Merge pull request #77 from storytold/feature/enqueue_tts_request

Feature/enqueue tts request
This commit is contained in:
Brandon Thomas authored and GitHub committed 2023-10-09 21:02:26 -04:00
commit b163791db2
19 files changed
+920 -86

No files matched your search

@@ -19,20 +19,18 @@ use strum::EnumIter;
pub enum InferenceModelType {
#[serde(rename = "rvc_v2")]
RvcV2,
// NB: sad_talker does use user-supplied models, so there is no "model token"
#[serde(rename = "sad_talker")]
SadTalker,
#[serde(rename = "so_vits_svc")]
SoVitsSvc,
// TODO: Does this need to be "legacy_tacotron2" ?
#[serde(rename = "tacotron2")]
Tacotron2,
#[serde(rename = "vits")]
Vits,
#[serde(rename = "vall_e_x")]
VallEX,
}
// TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
@@ -48,6 +46,7 @@ impl InferenceModelType {
Self::SoVitsSvc => "so_vits_svc",
Self::Tacotron2 => "tacotron2",
Self::Vits => "vits",
Self::VallEX => "vall_e_x"
}
}
@@ -58,6 +57,7 @@ impl InferenceModelType {
"so_vits_svc" => Ok(Self::SoVitsSvc),
"tacotron2" => Ok(Self::Tacotron2),
"vits" => Ok(Self::Vits),
"vall_e_x" => Ok(Self::VallEX),
_ => Err(format!("invalid value: {:?}", value)),
}
}
@@ -71,6 +71,7 @@ impl InferenceModelType {
InferenceModelType::SoVitsSvc,
InferenceModelType::Tacotron2,
InferenceModelType::Vits,
InferenceModelType::VallEX
])
}
}
@@ -87,6 +88,7 @@ mod tests {
assert_serialization(InferenceModelType::SoVitsSvc, "so_vits_svc");
assert_serialization(InferenceModelType::Tacotron2, "tacotron2");
assert_serialization(InferenceModelType::Vits, "vits");
assert_serialization(InferenceModelType::VallEX, "vall_e_x");
}
#[test]
@@ -96,6 +98,7 @@ mod tests {
assert_eq!(InferenceModelType::SoVitsSvc.to_str(), "so_vits_svc");
assert_eq!(InferenceModelType::Tacotron2.to_str(), "tacotron2");
assert_eq!(InferenceModelType::Vits.to_str(), "vits");
assert_eq!(InferenceModelType::VallEX.to_str(), "vall_e_x");
}
#[test]
@@ -105,18 +108,20 @@ mod tests {
assert_eq!(InferenceModelType::from_str("so_vits_svc").unwrap(), InferenceModelType::SoVitsSvc);
assert_eq!(InferenceModelType::from_str("tacotron2").unwrap(), InferenceModelType::Tacotron2);
assert_eq!(InferenceModelType::from_str("vits").unwrap(), InferenceModelType::Vits);
assert_eq!(InferenceModelType::from_str("vall_e_x").unwrap(), InferenceModelType::VallEX);
}
#[test]
fn all_variants() {
// Static check
let mut variants = InferenceModelType::all_variants();
assert_eq!(variants.len(), 5);
assert_eq!(variants.len(), 6);
assert_eq!(variants.pop_first(), Some(InferenceModelType::RvcV2));
assert_eq!(variants.pop_first(), Some(InferenceModelType::SadTalker));
assert_eq!(variants.pop_first(), Some(InferenceModelType::SoVitsSvc));
assert_eq!(variants.pop_first(), Some(InferenceModelType::Tacotron2));
assert_eq!(variants.pop_first(), Some(InferenceModelType::Vits));
assert_eq!(variants.pop_first(), Some(InferenceModelType::VallEX));
assert_eq!(variants.pop_first(), None);
// Generated check
@@ -2,6 +2,7 @@ use enums::by_table::generic_inference_jobs::inference_category::InferenceCatego
use errors::AnyhowResult;
use crate::payloads::generic_inference_args::lipsync_payload::LipsyncArgs;
use crate::payloads::generic_inference_args::tts_payload::TTSArgs;
/// Used to encode extra state for the `generic_inference_jobs` table.
/// This should act somewhat like a serialized protobuf stored inside a record.
@@ -52,10 +53,10 @@ pub enum PolymorphicInferenceArgs {
La (LipsyncArgs),
/// Text to speech. (Short name to save space when serializing.)
Tts {
Tts(TTSArgs)
// No arguments yet.
// It might be best to just not include this when not used.
},
,
/// Voice conversion. (Short name to save space when serializing.)
Vc {
/// Argument for so-vits-svc
@@ -142,12 +143,13 @@ mod tests {
assert!(json.len() < 1000);
}
// TODO FIX THIS TEST! MICHAEL
#[test]
fn typical_tts_args_serialize() {
let args = GenericInferenceArgs {
inference_category: Some(InferenceCategoryAbbreviated::VoiceConversion),
args: Some(PolymorphicInferenceArgs::Tts {
}),
args: Some(PolymorphicInferenceArgs::Tts()),
};
let json = serde_json::ser::to_string(&args).unwrap();
@@ -1,2 +1,3 @@
pub mod generic_inference_args;
pub mod lipsync_payload;
pub mod lipsync_payload;
pub mod tts_payload;
@@ -0,0 +1,17 @@
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct TTSArgs {
#[serde(rename = "t")]
pub text: String,
#[serde(rename = "e")]
pub voice_token: String // varchar 32
}
#[cfg(test)]
mod tests {
use crate::payloads::generic_inference_args::tts_payload::TTSArgs;
#[test]
fn test_tts_result() {
}
}
@@ -9,6 +9,18 @@ HTTP_BIND_ADDRESS=0.0.0.0:11223
# NB: Disable if not testing Redis
REDIS_FOR_KEEPALIVE_URL="redis://localhost/0"
# ================================= #
# #
# Vall-E-X #
# #
# ================================= #
# should rename my folder
# abs path to the folder that contains the inference code need to get that container in the repo need basepath? relative to everyones env
VALL_E_X_INFERENCE_ROOT_DIRECTORY = "/home/tensor/code/Storyteller/storyteller-ml/tts/VALL-E-X"
VALL_E_X_INFERENCE_COMMAND = "python3 main.py"
VALL_E_X_INFERENCE_MAYBE_VENV_COMMAND = "source .env/bin/activate"
VALL_E_X_INFERENCE_MAYBE_DOCKER_IMAGE = "" # fill this out for dev?
# ================================= #
# #
# RVC (v2) #
@@ -84,7 +84,7 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
.new_tempdir(&work_temp_dir)
.map_err(|e| ProcessSingleJobError::from_io_error(e))?;
// ==================== QUERY AND DOWNLOAD FILES ==================== //
let audio_path = download_audio_file(
@@ -178,9 +178,7 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
// ==================== RUN INFERENCE SCRIPT ==================== //
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
@@ -1,3 +1,4 @@
pub mod process_single_tts_job;
pub mod tacotron2_v2_early_fakeyou;
pub mod vits;
pub mod vall_e_x;
@@ -1,5 +1,6 @@
use anyhow::anyhow;
use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use enums::by_table::tts_models::tts_model_type::TtsModelType;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::tts::tts_models::get_tts_model_for_inference_improved::get_tts_model_for_inference_improved;
@@ -7,80 +8,110 @@ use tokens::tokens::tts_models::TtsModelToken;
use crate::job::job_loop::job_success_result::JobSuccessResult;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::tts::{tacotron2_v2_early_fakeyou, vits};
use crate::job::job_types::tts::{ tacotron2_v2_early_fakeyou, vits };
use crate::job::job_types::tts::tacotron2_v2_early_fakeyou::process_job::ProcessJobArgs;
use crate::job::job_types::tts::vits::process_job::VitsProcessJobArgs;
use crate::job::job_types::tts::vall_e_x::process_job::VALLEXProcessJobArgs;
use crate::job_dependencies::JobDependencies;
pub async fn process_single_tts_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result<JobSuccessResult, ProcessSingleJobError> {
use super::vall_e_x;
let tts_model_token = match job.maybe_model_token.as_deref() {
None => return Err(ProcessSingleJobError::Other(anyhow!("no model token on job"))),
Some(model_token) => TtsModelToken::new_from_str(model_token),
};
pub async fn process_single_tts_job(
job_dependencies: &JobDependencies,
job: &AvailableInferenceJob
) -> Result<JobSuccessResult, ProcessSingleJobError> {
// TODO: Move common checks for slurs, etc. here.
let tts_model_token = match job.maybe_model_token.as_deref() {
None => {
return Err(ProcessSingleJobError::Other(anyhow!("no model token on job")));
}
Some(model_token) => TtsModelToken::new_from_str(model_token),
};
let raw_inference_text = job.maybe_raw_inference_text
.as_deref()
.ok_or(ProcessSingleJobError::Other(anyhow!("no inference text")))?;
// TODO: Move common checks for slurs, etc. here.
let raw_inference_text = job.maybe_raw_inference_text
.as_deref()
.ok_or(ProcessSingleJobError::Other(anyhow!("no inference text")))?;
// TODO: Interrogate cache (which also depends on other flags)
let maybe_tts_model = get_tts_model_for_inference_improved(
&job_dependencies.mysql_pool, tts_model_token.as_str())
.await
.map_err(|err| ProcessSingleJobError::Other(anyhow!("database error: {:?}", err)))?;
// TODO: Interrogate cache (which also depends on other flags)
let maybe_tts_model = get_tts_model_for_inference_improved(
&job_dependencies.mysql_pool,
tts_model_token.as_str()
).await.map_err(|err| ProcessSingleJobError::Other(anyhow!("database error: {:?}", err)))?;
// TODO: Attempt to grab job lock
// TODO: Attempt to grab job lock
// // ==================== ATTEMPT TO GRAB JOB LOCK ==================== //
//
// info!("Attempting to grab lock for job: {}", job.inference_job_token);
//
// let lock_acquired =
// mark_tts_inference_job_pending_and_grab_lock(&job_args.mysql_pool, job.id)
// .await
// .map_err(|e| ProcessSingleJobError::Other(e))?;
//
// if !lock_acquired {
// warn!("Could not acquire job lock for: {:?}", &job.id);
// let duration = start.elapsed();
//
// since_creation_span.set_attribute("status", "failure");
// since_creation_span.set_duration(duration);
//
// job_iteration_span.set_attribute("status", "failure");
// job_iteration_span.set_duration(duration);
//
// return Ok((since_creation_span, job_iteration_span));
// }
//
// info!("Lock acquired for job: {}", job.inference_job_token);
// // ==================== ATTEMPT TO GRAB JOB LOCK ==================== //
//
// info!("Attempting to grab lock for job: {}", job.inference_job_token);
//
// let lock_acquired =
// mark_tts_inference_job_pending_and_grab_lock(&job_args.mysql_pool, job.id)
// .await
// .map_err(|e| ProcessSingleJobError::Other(e))?;
//
// if !lock_acquired {
// warn!("Could not acquire job lock for: {:?}", &job.id);
// let duration = start.elapsed();
//
// since_creation_span.set_attribute("status", "failure");
// since_creation_span.set_duration(duration);
//
// job_iteration_span.set_attribute("status", "failure");
// job_iteration_span.set_duration(duration);
//
// return Ok((since_creation_span, job_iteration_span));
// }
//
// info!("Lock acquired for job: {}", job.inference_job_token);
// Look for the Zeroshot job type
// let job_success_result = match job.maybe_model_type {
// Some(InferenceModelType::VallEX) => {
let tts_model = match maybe_tts_model {
None => return Err(ProcessSingleJobError::Other(anyhow!("tts model not found: {:?}", tts_model_token))),
Some(model) => model,
};
// }
// Some(model_type) => return Err(ProcessSingleJobError::Other(anyhow!("wrong model type: {:?}", model_type))),
// None => return Err(ProcessSingleJobError::Other(anyhow!("no model type in record"))),
// };
let job_success_result = match tts_model.tts_model_type {
TtsModelType::Tacotron2 => {
tacotron2_v2_early_fakeyou::process_job::process_job(ProcessJobArgs {
job_dependencies,
job,
tts_model: &tts_model,
raw_inference_text,
}).await?
if let Some(model_type) = job.maybe_model_type {
if model_type == InferenceModelType::VallEX {
let job_success_result = vall_e_x::process_job::process_job(VALLEXProcessJobArgs {
job_dependencies,
job,
}).await?;
Ok::<JobSuccessResult,ProcessSingleJobError>(job_success_result);
}
// do nothing try regular tts else other zero shot models
}
TtsModelType::Vits => {
vits::process_job::process_job(VitsProcessJobArgs {
job_dependencies,
job,
tts_model: &tts_model,
raw_inference_text,
}).await?
}
};
Ok(job_success_result)
// lets run the regular tts
let tts_model = match maybe_tts_model {
None => {
return Err(
ProcessSingleJobError::Other(anyhow!("tts model not found: {:?}", tts_model_token))
);
}
Some(model) => model,
};
let job_success_result: JobSuccessResult = match tts_model.tts_model_type {
TtsModelType::Tacotron2 => {
tacotron2_v2_early_fakeyou::process_job::process_job(ProcessJobArgs {
job_dependencies,
job,
tts_model: &tts_model,
raw_inference_text,
}).await?
}
TtsModelType::Vits => {
vits::process_job::process_job(VitsProcessJobArgs {
job_dependencies,
job,
tts_model: &tts_model,
raw_inference_text,
}).await?
}
};
Ok(job_success_result)
}
@@ -0,0 +1,5 @@
pub mod model_downloaders;
pub mod vall_e_x_inference_command;
pub mod process_job;
pub mod validate_job;
pub mod download_embedding_file;
@@ -0,0 +1,91 @@
use crate::util::model_downloader::ModelDownloader;
pub struct VallEXDownloaders {
pub vall_e_x: VallEXModelMapping,
pub whisper_lg: WhisperLargeModelMapping,
pub whisper_md: WhisperMediumModelMapping,
pub vocos_encodec_24khz: VocosEncodecModelMapping
}
crate::impl_model_downloader!(
VallEXModelMapping,
"Vall-e-x Model Mapping",
"VALL_E_X_MAPPING_BUCKET_PATH",
"/dependencies/zero_shot_tts/vall-e-x_1.0",
"VALL_E_X_MAPPING_FILESYSTEM_PATH",
"/tmp/downloads/zero_shot_tts/vall-e-x_1.0"
);
crate::impl_model_downloader!(
WhisperLargeModelMapping,
"whisper Large Model Mapping",
"WHISPER_LARGE_MAPPING_BUCKET_PATH",
"dependencies/zero_shot_tts/vall-e-x_1.0/whisper-large",
"WHISPER_LARGE_MAPPING_FILESYSTEM_PATH",
"/tmp/downloads/zero_shot_tts/vall-e-x_1.0/whisper-large"
);
crate::impl_model_downloader!(
WhisperMediumModelMapping,
"whisper Medium Model Mapping",
"WHISPER_MEDIUM_TENSOR_BUCKET_PATH",
"dependencies/zero_shot_tts/vall-e-x_1.0/whisper-medium",
"WHISPER_MEDIUM_TENSOR_FILESYSTEM_PATH",
"/tmp/downloads/zero_shot_tts/vall-e-x_1.0/whisper-medium"
);
crate::impl_model_downloader!(
VocosEncodecModelMapping,
"Vocos Encodec Model Mapping",
"VOCOS_ENCODEC_TENSOR_BUCKET_PATH",
"dependencies/zero_shot_tts/vall-e-x_1.0/encodec_pytorch_model.bin",
"VOCOS_ENCODEC_TENSOR_FILESYSTEM_PATH",
"/tmp/downloads/zero_shot_tts/vall-e-x_1.0/encodec_pytorch_model.bin"
);
impl VallEXDownloaders {
pub fn build_all_from_env() -> Self {
Self {
vall_e_x: VallEXModelMapping::from_env(),
whisper_lg: WhisperLargeModelMapping::from_env(),
whisper_md: WhisperMediumModelMapping::from_env(),
vocos_encodec_24khz: VocosEncodecModelMapping::from_env()
}
}
pub fn all_downloaders(&self) -> Vec<&dyn ModelDownloader> {
vec![
&self.vall_e_x,
&self.whisper_lg,
&self.whisper_md,
&self.vocos_encodec_24khz,
]
}
}
mod test {
// use std::path::Path;
// use crate::job::job_types::lipsync::sad_talker::model_downloaders::{SadTalkerModelFirstMapping, SadTalkerModelSecondMapping};
// use crate::util::model_downloader::ModelDownloader;
// #[test]
// fn test_downloader_model_first_mapping() {
// let model = SadTalkerModelFirstMapping::default();
// assert_eq!(model.get_model_name(), "sad talker model first mapping");
// assert_eq!(model.get_cloud_bucket_path(), "/animation_sadtalker/model/mapping_00109-model.pth.tar");
// assert_eq!(model.get_filesystem_path(), Path::new("/tmp/downloads/sadtalker/mapping_00109-model.pth.tar"));
// }
// #[test]
// fn test_downloader_model_second_mapping() {
// let model = SadTalkerModelSecondMapping::default();
// assert_eq!(model.get_model_name(), "sad talker model second mapping");
// assert_eq!(model.get_cloud_bucket_path(), "/animation_sadtalker/model/mapping_00229-model.pth.tar");
// assert_eq!(model.get_filesystem_path(), Path::new("/tmp/downloads/sadtalker/mapping_00229-model.pth.tar"));
// }
}
@@ -0,0 +1,98 @@
use std::fs::read_to_string;
use std::time::Instant;
use anyhow::anyhow;
use log::{error, info, warn};
use buckets::public::media_files::original_file::MediaFileBucketPath;
use container_common::filesystem::check_file_exists::check_file_exists;
use container_common::filesystem::safe_delete_temp_directory::safe_delete_temp_directory;
use container_common::filesystem::safe_delete_temp_file::safe_delete_temp_file;
use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
use filesys::file_size::file_size;
use hashing::sha256::sha256_hash_file::sha256_hash_file;
use mimetypes::mimetype_for_file::get_mimetype_for_file;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::media_files::insert_media_file_from_face_animation::{insert_media_file_from_face_animation, InsertArgs};
use tokens::users::user::UserToken;
use crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
//use crate::job::job_types::lipsync::sad_talker::download_audio_file::download_audio_file;
//use crate::job::job_types::lipsync::sad_talker::sad_talker_inference_command::InferenceArgs;
use crate::job::job_types::tts::vall_e_x::validate_job::validate_job;
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 async fn process_job(args: VALLEXProcessJobArgs<'_>) -> Result<JobSuccessResult, ProcessSingleJobError> {
// get job args and dependencies and execute them with the inputs
let job = args.job;
let deps = args.job_dependencies;
let mut 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)))?;
// validate the inputs on the job shouldn't this happen at json
//let job_args = validate_job(job)?;
// Need to download the models
info!("Download models (if not present)...");
let mut i : usize = 0;
for downloader in deps.job_type_details.vall_e_x.downloaders.all_downloaders() {
// Temporary debugging
info!("Downloader {}", i);
i = i + 1;
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
// ==================== TEMP DIR ==================== //
// create a temp dir
// run inference
// upload audio to bucket
// deletetemp dir files
// save result ?
}
#[cfg(test)]
mod test {
#[test]
fn values() {
print!("{}","Hello!");
}
}
@@ -0,0 +1,9 @@
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use crate::job::{job_types::lipsync::sad_talker::validate_job::JobArgs, job_loop::process_single_job_error::ProcessSingleJobError};
pub fn validate_job(job: &AvailableInferenceJob) -> Result<JobArgs,ProcessSingleJobError> {
}
@@ -0,0 +1,369 @@
use std::collections::HashSet;
use std::env;
use std::ffi::OsString;
use std::fs::File;
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::anyhow;
use log::info;
use once_cell::sync::Lazy;
use subprocess::{Popen, PopenConfig, Redirection};
use container_common::anyhow_result::AnyhowResult;
use filesys::path_to_string::path_to_string;
use subprocess_common::docker_options::{DockerFilesystemMount, DockerGpu, DockerOptions};
use crate::job::job_loop::command_exit_status::CommandExitStatus;
#[derive(Clone)]
pub enum ExecutableOrCommand {
/// Eg. `inference.py`
Executable(PathBuf),
/// Eg. `python3 inference.py`
Command(String),
}
// These environment vars are not copied over to the subprocess
// TODO/FIXME(bt, 2023-05-28): This is horrific security!
static IGNORED_ENVIRONMENT_VARS : Lazy<HashSet<String>> = Lazy::new(|| {
let env_var_names= [
"MYSQL_URL",
"ACCESS_KEY",
"SECRET_KEY",
"NEWRELIC_API_KEY",
];
env_var_names.iter()
.map(|value| value.to_string())
.collect::<HashSet<String>>()
});
// The inference command assumes everything has been setup to run inference
// 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_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
pub stderr_output_file: P
}
#[derive(Clone)]
pub struct VallEXInferenceCommand {
/// Where the code lives
root_code_directory: PathBuf,
/// A single executable script or a much larger bash command.
executable_or_command: ExecutableOrCommand,
/// eg. `source python/bin/activate`
maybe_virtual_env_activation_command: Option<String>,
/// Optional default config file to use
maybe_default_config_path: Option<PathBuf>,
/// If this is run under Docker (eg. in development), these are the options.
maybe_docker_options: Option<DockerOptions>,
/// If the execution should be ended after a certain point.
maybe_execution_timeout: Option<Duration>,
/// Inference arg.
/// --checkpoint_dir: optional location for checkpoints directory
pub alternate_checkpoint_dir: Option<PathBuf>
}
impl VallEXInferenceCommand {
pub fn new<P: AsRef<Path>>(
root_code_directory: P,
executable_or_command: ExecutableOrCommand,
maybe_virtual_env_activation_command: Option<&str>,
maybe_default_config_path: Option<P>,
maybe_docker_options: Option<DockerOptions>,
maybe_execution_timeout: Option<Duration>,
alternate_checkpoint_dir: Option<PathBuf>
) -> Self {
Self {
root_code_directory: root_code_directory.as_ref().to_path_buf(),
executable_or_command,
maybe_virtual_env_activation_command: maybe_virtual_env_activation_command.map(|s| s.to_string()),
maybe_default_config_path: maybe_default_config_path.map(|p| p.as_ref().to_path_buf()),
maybe_docker_options,
maybe_execution_timeout,
alternate_checkpoint_dir,
}
}
pub fn from_env() -> AnyhowResult<Self> {
let root_code_directory = easyenv::get_env_pathbuf_required(
"VALL_E_X_INFERENCE_ROOT_DIRECTORY")?;
let maybe_inference_command = easyenv::get_env_string_optional(
"VALL_E_X_INFERENCE_COMMAND");
// EVENTUALLY should remove the check
// Optional, eg. `./infer.py`. Typically we'll use the command form instead.
let maybe_inference_executable = easyenv::get_env_pathbuf_optional(
"VALL_E_X_INFERENCE_EXECUTABLE");
let executable_or_command = match maybe_inference_command {
Some(command) => ExecutableOrCommand::Command(command),
None => match maybe_inference_executable {
Some(executable) => ExecutableOrCommand::Executable(executable),
None => return Err(anyhow!("neither command nor executable passed")),
},
};
let maybe_virtual_env_activation_command = easyenv::get_env_string_optional(
"VALL_E_X_INFERENCE_MAYBE_VENV_COMMAND");
let maybe_default_config_path = easyenv::get_env_pathbuf_optional(
"VALL_E_X_INFERENCE_MAYBE_DEFAULT_CONFIG_PATH");
let maybe_execution_timeout =
easyenv::get_env_duration_seconds_optional("TIMEOUT_SECONDS");
// Probably for local
let maybe_docker_options = easyenv::get_env_string_optional(
"VALL_E_X_INFERENCE_MAYBE_DOCKER_IMAGE")
.map(|image_name| {
DockerOptions {
image_name,
maybe_bind_mount: Some(DockerFilesystemMount::tmp_to_tmp()),
maybe_environment_variables: None,
maybe_gpu: Some(DockerGpu::All),
}
});
// Override for --checkpoint_dir at inference time
let alternate_checkpoint_dir = easyenv::get_env_pathbuf_optional(
"SAD_TALKER_ALTERNATE_CHECKPOINT_PATH");
Ok(Self {
root_code_directory,
executable_or_command,
maybe_virtual_env_activation_command,
maybe_default_config_path,
maybe_docker_options,
maybe_execution_timeout,
alternate_checkpoint_dir,
})
}
pub fn execute_inference<P: AsRef<Path>>(
&self,
args: InferenceArgs<P>,
) -> CommandExitStatus {
match self.do_execute_inference(args) {
Ok(exit_status) => exit_status,
Err(error) => CommandExitStatus::FailureWithReason { reason: format!("error: {:?}", error) },
}
}
fn do_execute_inference<P: AsRef<Path>>(
&self,
args: InferenceArgs<P>,
) -> AnyhowResult<CommandExitStatus> {
let mut command = String::new();
command.push_str(&format!("cd {}", path_to_string(&self.root_code_directory)));
if let Some(venv_command) = self.maybe_virtual_env_activation_command.as_deref() {
command.push_str(" && ");
command.push_str(venv_command);
command.push_str(" ");
}
command.push_str(" && ");
match self.executable_or_command {
ExecutableOrCommand::Executable(ref executable) => {
command.push_str(&path_to_string(executable));
command.push_str(" ");
}
ExecutableOrCommand::Command(ref cmd) => {
command.push_str(cmd);
command.push_str(" ");
}
}
// ===== Begin Python Args =====
command.push_str(" -text ");
command.push_str(&path_to_string(args.input_text));
command.push_str(" -use_prompt_name ");
command.push_str(&path_to_string(args.input_embedding));
command.push_str(" -output_file_name ");
command.push_str(&path_to_string(args.output_file_name));
// ===== End Python Args =====
if let Some(docker_options) = self.maybe_docker_options.as_ref() {
command = docker_options.to_command_string(&command);
}
info!("Command: {:?}", command);
let command_parts = [
"bash",
"-c",
&command
];
let mut env_vars = Vec::new();
// Copy all environment variables from the parent process.
// This is necessary to send all the kubernetes settings for Nvidia / CUDA.
for (env_key, env_value) in env::vars() {
if IGNORED_ENVIRONMENT_VARS.contains(&env_key) {
continue;
}
env_vars.push((
OsString::from(env_key),
OsString::from(env_value),
));
}
let mut config = PopenConfig::default();
info!("stderr will be written to file: {:?}", args.stderr_output_file.as_ref());
let stderr_file = File::create(&args.stderr_output_file)?;
config.stderr = Redirection::File(stderr_file);
if !env_vars.is_empty() {
config.env = Some(env_vars);
}
let mut p = Popen::create(&command_parts, config)?;
info!("Subprocess PID: {:?}", p.pid());
match self.maybe_execution_timeout {
None => {
let exit_status = p.wait()?;
info!("Subprocess exit status: {:?}", exit_status);
Ok(CommandExitStatus::from_exit_status(exit_status))
}
Some(timeout) => {
info!("Executing with timeout: {:?}", &timeout);
let exit_status = p.wait_timeout(timeout.clone())?;
match exit_status {
None => {
// NB: If the program didn't successfully terminate, kill it.
info!("Subprocess didn't end after timeout: {:?}; terminating...", &timeout);
let _r = p.terminate()?;
Ok(CommandExitStatus::Timeout)
}
Some(exit_status) => {
info!("Subprocess timed wait exit status: {:?}", exit_status);
Ok(CommandExitStatus::from_exit_status(exit_status))
}
}
}
}
}
}
#[derive(Clone)]
pub struct VallEXCreateEmbeddingCommand {
/// Where the code lives
root_code_directory: PathBuf,
/// A single executable script or a much larger bash command.
executable_or_command: ExecutableOrCommand,
/// eg. `source python/bin/activate`
maybe_virtual_env_activation_command: Option<String>,
/// Optional default config file to use
maybe_default_config_path: Option<PathBuf>,
/// If this is run under Docker (eg. in development), these are the options.
maybe_docker_options: Option<DockerOptions>,
/// If the execution should be ended after a certain point.
maybe_execution_timeout: Option<Duration>,
/// Inference arg.
/// --checkpoint_dir: optional location for checkpoints directory
pub alternate_checkpoint_dir: Option<PathBuf>
}
impl VallEXCreateEmbeddingCommand {
pub fn new<P: AsRef<Path>>(
root_code_directory: P,
executable_or_command: ExecutableOrCommand,
maybe_virtual_env_activation_command: Option<&str>,
maybe_default_config_path: Option<P>,
maybe_docker_options: Option<DockerOptions>,
maybe_execution_timeout: Option<Duration>,
alternate_checkpoint_dir: Option<PathBuf>
) -> Self {
Self {
root_code_directory: root_code_directory.as_ref().to_path_buf(),
executable_or_command,
maybe_virtual_env_activation_command: maybe_virtual_env_activation_command.map(|s| s.to_string()),
maybe_default_config_path: maybe_default_config_path.map(|p| p.as_ref().to_path_buf()),
maybe_docker_options,
maybe_execution_timeout,
alternate_checkpoint_dir,
}
}
// helper funciton to set up env
pub fn from_env() -> AnyhowResult<Self> {
// TODO: fix for now just for compilation
let root_code_directory = easyenv::get_env_pathbuf_required(
"VALL_E_X_INFERENCE_ROOT_DIRECTORY")?;
let maybe_inference_command = easyenv::get_env_string_optional(
"VALL_E_X_INFERENCE_COMMAND");
// EVENTUALLY should remove the check
// Optional, eg. `./infer.py`. Typically we'll use the command form instead.
let maybe_inference_executable = easyenv::get_env_pathbuf_optional(
"VALL_E_X_INFERENCE_EXECUTABLE");
let executable_or_command = match maybe_inference_command {
Some(command) => ExecutableOrCommand::Command(command),
None => match maybe_inference_executable {
Some(executable) => ExecutableOrCommand::Executable(executable),
None => return Err(anyhow!("neither command nor executable passed")),
},
};
let maybe_virtual_env_activation_command = easyenv::get_env_string_optional(
"VALL_E_X_INFERENCE_MAYBE_VENV_COMMAND");
let maybe_default_config_path = easyenv::get_env_pathbuf_optional(
"VALL_E_X_INFERENCE_MAYBE_DEFAULT_CONFIG_PATH");
let maybe_execution_timeout =
easyenv::get_env_duration_seconds_optional("TIMEOUT_SECONDS");
// Probably for local
let maybe_docker_options = easyenv::get_env_string_optional(
"VALL_E_X_INFERENCE_MAYBE_DOCKER_IMAGE")
.map(|image_name| {
DockerOptions {
image_name,
maybe_bind_mount: Some(DockerFilesystemMount::tmp_to_tmp()),
maybe_environment_variables: None,
maybe_gpu: Some(DockerGpu::All),
}
});
// Override for --checkpoint_dir at inference time
let alternate_checkpoint_dir = easyenv::get_env_pathbuf_optional(
"SAD_TALKER_ALTERNATE_CHECKPOINT_PATH");
Ok(Self {
root_code_directory,
executable_or_command,
maybe_virtual_env_activation_command,
maybe_default_config_path,
maybe_docker_options,
maybe_execution_timeout,
alternate_checkpoint_dir,
})
}
}
@@ -22,7 +22,12 @@ use newrelic_telemetry::Client as NewRelicClient;
use crate::job::job_types::lipsync::sad_talker::model_downloaders::SadTalkerDownloaders;
use crate::job::job_types::lipsync::sad_talker::sad_talker_inference_command::SadTalkerInferenceCommand;
use crate::job::job_types::tts::tacotron2_v2_early_fakeyou::tacotron2_inference_command::Tacotron2InferenceCommand;
use crate::job::job_types::tts::vall_e_x::model_downloaders::VallEXDownloaders;
use crate::job::job_types::tts::vits::vits_inference_command::VitsInferenceCommand;
use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXInferenceCommand;
use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXCreateEmbeddingCommand;
use crate::job::job_types::vc::rvc_v2::pretrained_hubert_model::PretrainedHubertModel;
use crate::job::job_types::vc::rvc_v2::rvc_v2_inference_command::RvcV2InferenceCommand;
use crate::job::job_types::vc::so_vits_svc::so_vits_svc_inference_command::SoVitsSvcInferenceCommand;
@@ -161,6 +166,7 @@ pub struct JobTypeDetails {
pub rvc_v2: RvcV2Details,
pub so_vits_svc: SoVitsSvcDetails,
pub sad_talker: SadTalkerDetails,
pub vall_e_x: VallEXDetails
//pub tacotron2_modern: ...,
//pub softvc: ...,
}
@@ -196,3 +202,11 @@ pub struct SadTalkerDetails {
pub inference_command: SadTalkerInferenceCommand,
pub ffmpeg_watermark_command: FfmpegLogoWatermarkCommand,
}
// TODO: we will probably want a command type of some kind that implements
// some kind of interface that we can just pass strings to.
pub struct VallEXDetails {
pub downloaders: VallEXDownloaders,
pub inference_command: VallEXInferenceCommand,
pub create_embedding_command: VallEXCreateEmbeddingCommand
}
+13 -1
View File
@@ -53,11 +53,18 @@ use crate::job::job_loop::main_loop::main_loop;
use crate::job::job_types::lipsync::sad_talker::model_downloaders::SadTalkerDownloaders;
use crate::job::job_types::lipsync::sad_talker::sad_talker_inference_command::SadTalkerInferenceCommand;
use crate::job::job_types::tts::tacotron2_v2_early_fakeyou::tacotron2_inference_command::Tacotron2InferenceCommand;
use crate::job::job_types::tts::vits::vits_inference_command::VitsInferenceCommand;
use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXInferenceCommand;
use crate::job::job_types::tts::vall_e_x::vall_e_x_inference_command::VallEXCreateEmbeddingCommand;
use crate::job::job_types::tts::vall_e_x::model_downloaders::VallEXDownloaders;
use crate::job::job_types::vc::rvc_v2::pretrained_hubert_model::PretrainedHubertModel;
use crate::job::job_types::vc::rvc_v2::rvc_v2_inference_command::RvcV2InferenceCommand;
use crate::job::job_types::vc::so_vits_svc::so_vits_svc_inference_command::SoVitsSvcInferenceCommand;
use crate::job_dependencies::{FileSystemDetails, JobCaches, JobDependencies, JobTypeDetails, JobWorkerDetails, PretrainedModels, RvcV2Details, SadTalkerDetails, SoVitsSvcDetails, Tacotron2VocodesDetails, VitsDetails};
use crate::job_dependencies::{FileSystemDetails, JobCaches, JobDependencies, JobTypeDetails, JobWorkerDetails, PretrainedModels, RvcV2Details, SadTalkerDetails, SoVitsSvcDetails, Tacotron2VocodesDetails, VitsDetails, VallEXDetails};
use crate::util::common_commands::ffmpeg_logo_watermark_command::FfmpegLogoWatermarkCommand;
use crate::util::scoped_execution::ScopedExecution;
use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator;
@@ -323,6 +330,11 @@ async fn main() -> AnyhowResult<()> {
hifigan_vocoder_model_filename,
hifigan_superres_vocoder_model_filename,
},
vall_e_x: VallEXDetails {
downloaders: VallEXDownloaders::build_all_from_env(),
inference_command: VallEXInferenceCommand::from_env()?,
create_embedding_command: VallEXCreateEmbeddingCommand::from_env()?,
},
vits: VitsDetails {
inference_command: vits_inference_command()?,
},
@@ -227,7 +227,7 @@ pub async fn enqueue_lipsync_animation_handler(
// TODO(bt): CHECK DATABASE FOR TOKENS!
let audio_source = {
let audio_source: LipsyncAnimationAudioSource = {
if let Some(ref token) = request.audio_source.maybe_media_file_token {
LipsyncAnimationAudioSource::media_file_token(token.as_str())
} else if let Some(ref token) = request.audio_source.maybe_media_upload_token {
@@ -241,7 +241,7 @@ pub async fn enqueue_lipsync_animation_handler(
}
};
let image_source = {
let image_source: LipsyncAnimationImageSource = {
if let Some(ref token) = request.image_source.maybe_media_file_token {
LipsyncAnimationImageSource::media_file_token(token.as_str())
} else if let Some(ref token) = request.image_source.maybe_media_upload_token {
@@ -255,7 +255,7 @@ pub async fn enqueue_lipsync_animation_handler(
let maybe_user_preferred_visibility : Option<Visibility> = maybe_user_session
.as_ref()
.map(|user_session| user_session.preferences.preferred_tts_result_visibility); // TODO: New setting for web-vc
.map(|user_session: &users_component::utils::user_session_extended::UserSessionExtended| user_session.preferences.preferred_tts_result_visibility); // TODO: New setting for web-vc
let set_visibility = request.creator_set_visibility
.or(maybe_user_preferred_visibility)
@@ -329,6 +329,8 @@ pub async fn enqueue_lipsync_animation_handler(
}
};
// If you are using this file as a reference for another generic endpoint feature.
// this feature will be sunset so don't worry about this
server_state.firehose_publisher.enqueue_lipsync_animation(
maybe_user_token.as_ref(),
&job_token)
@@ -338,7 +340,7 @@ pub async fn enqueue_lipsync_animation_handler(
EnqueueLipsyncAnimationError::ServerError
})?;
let response = EnqueueLipsyncAnimationSuccessResponse {
let response: EnqueueLipsyncAnimationSuccessResponse = EnqueueLipsyncAnimationSuccessResponse {
success: true,
inference_job_token: job_token,
};
@@ -1,15 +1,181 @@
#![forbid(unused_imports)]
#![forbid(unused_mut)]
#![forbid(unused_variables)]
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use mysql_queries::payloads::generic_inference_args::generic_inference_args::{GenericInferenceArgs, InferenceCategoryAbbreviated, PolymorphicInferenceArgs};
use std::sync::Arc;
use actix_web::{HttpRequest, HttpResponse, web};
use actix_web::error::ResponseError;
use actix_web::http::StatusCode;
use log::{warn};
use http_server_common::request::get_request_ip::get_request_ip;
use log::{info, warn};
use tokens::users::user::UserToken;
use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use mysql_queries::payloads::generic_inference_args::tts_payload::TTSArgs;
use crate::http_server::web_utils::response_error_helpers::to_simple_json_error;
use tokens::jobs::inference::InferenceJobToken;
use crate::server_state::ServerState;
use mysql_queries::queries::generic_inference::web::insert_generic_inference_job::{insert_generic_inference_job, InsertGenericInferenceArgs};
// TODO MOVE into own file.
use std::fmt::Debug;
use serde::Deserialize;
use serde::Serialize;
#[derive(Deserialize)]
pub struct EnqueueTTSRequest {
uuid_idempotency_token: String,
text: String,
embedding_token: String
}
#[derive(Serialize)]
pub struct EnqueueTTSRequestSuccessResponse {
pub success: bool,
pub inference_job_token: InferenceJobToken
}
#[derive(Debug)]
pub enum EnqueueTTSRequestError {
BadInput(String),
NotAuthorized,
ServerError,
RateLimited,
}
impl ResponseError for EnqueueTTSRequestError {
fn status_code(&self) -> StatusCode {
match *self {
EnqueueTTSRequestError::BadInput(_) => StatusCode::BAD_REQUEST,
EnqueueTTSRequestError::NotAuthorized => StatusCode::UNAUTHORIZED,
EnqueueTTSRequestError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
EnqueueTTSRequestError::RateLimited => StatusCode::TOO_MANY_REQUESTS,
}
}
fn error_response(&self) -> HttpResponse {
let error_reason = match self {
EnqueueTTSRequestError::BadInput(reason) => reason.to_string(),
EnqueueTTSRequestError::NotAuthorized => "unauthorized".to_string(),
EnqueueTTSRequestError::ServerError => "server error".to_string(),
EnqueueTTSRequestError::RateLimited => "rate limited".to_string(),
};
to_simple_json_error(&error_reason, self.status_code())
}
}
impl std::fmt::Display for EnqueueTTSRequestError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?}", self)
}
}
// Implementation for enqueuing a TTS request
// Reference enqueue_infer_tts_handler.rs for checks: rate limiting / user sessions
// insert generic inference job.rs
// Need to convert it to generic inference job.
pub async fn enqueue_tts_request(
http_request: HttpRequest,
request: web::Json<EnqueueTTSRequest>,
server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse,EnqueueTTSRequestError> {
println!("Recieved payload");
let is_debug_request = true;
let maybe_user_token : Option<UserToken> = Some(UserToken::new_from_str(&"place holder")); // TODO fix this
let priority_level = 0;
//let disable_rate_limiter = false; // NB: Careful!
// do something with user session check if the user should even be able to access the end point
// CAN address this soon just getting everything out of the way to just enqueue the job.
// GET MY SQL
// let mut mysql_connection = server_state.mysql_pool
// .acquire()
// .await
// .map_err(|err| {
// warn!("MySql pool error: {:?}", err);
// EnqueueTTSRequestError::ServerError
// })?;
// return errors on payload not being correct
//Err(EnqueueTTSRequestError::ServerError) # missing voice token
// TODO: check for session
// TODO: Give investors priority
// TODO: Add Rate Limiter
// TODO: Look up model info?
// TODO(bt): CHECK DATABASE FOR TOKENS? I am guessing we need to ensure those tokens exist because the files may not uploaded or are availible?
// get input from the container spec and create a object that similar to llipsync_payload.rs
// PACKAGE JSON into RUST Struct, any smaller components
// remap from the request
// pass tokens from the request and create a payload that will have the information.
// check malformed json
// Get up IP address
let ip_address = get_request_ip(&http_request);
// package as larger component args should always have an embedding token ..
let inference_args = TTSArgs {
text: request.text.clone(),
voice_token: request.embedding_token.clone()
};
// create the inference args here
// enqueue a zero shot tts request here...
// create the job record here! explore the table insert the inference args in here as json! keep it short
let query_result = insert_generic_inference_job(InsertGenericInferenceArgs {
uuid_idempotency_token: &request.uuid_idempotency_token,
inference_category: InferenceCategory::TextToSpeech,
maybe_model_type: Some(InferenceModelType::VallEX), // NB: Model is static during inference
maybe_model_token: None, // NB: Model is static during inference
maybe_input_source_token: None, // TODO: Introduce a second foreign key ?
maybe_input_source_token_type: None, // TODO: Introduce a second foreign key ?
maybe_raw_inference_text: None, // No text
maybe_inference_args: Some(GenericInferenceArgs {
inference_category: Some(InferenceCategoryAbbreviated::TextToSpeech),
args: Some(PolymorphicInferenceArgs::Tts(inference_args)),
}),
maybe_creator_user_token: maybe_user_token.as_ref(),
creator_ip_address: &ip_address,
creator_set_visibility: enums::common::visibility::Visibility::Public,
priority_level,
requires_keepalive: true, // do we need this? I think so
is_debug_request,
maybe_routing_tag: Some("Tag"), // TODO fix later what is this tag for?
mysql_pool: &server_state.mysql_pool,
}).await;
let job_token = match query_result {
Ok((job_token, _id)) => job_token,
Err(err) => {
warn!("New generic inference job creation DB error: {:?}", err);
return Err(EnqueueTTSRequestError::ServerError);
}
};
let response: EnqueueTTSRequestSuccessResponse = EnqueueTTSRequestSuccessResponse {
success: true,
inference_job_token: job_token,
};
let body = serde_json::to_string(&response)
.map_err(|_e| EnqueueTTSRequestError::ServerError)?;
// Error handling 101 rust result type returned like so.
Ok(HttpResponse::Ok()
.content_type("application/json")
.body(body))
pub async fn enqueue_tts_request() -> HttpResponse {
// Implementation for enqueuing a TTS request
HttpResponse::Ok().json("TTS request enqueued successfully")
}
@@ -75,6 +75,7 @@ use crate::http_server::endpoints::tts::delete_tts_model::delete_tts_model_handl
use crate::http_server::endpoints::tts::delete_tts_result::delete_tts_inference_result_handler;
use crate::http_server::endpoints::tts::edit_tts_model::edit_tts_model_handler;
use crate::http_server::endpoints::tts::edit_tts_result::edit_tts_inference_result_handler;
use crate::http_server::endpoints::tts::enqueue_infer_tts_handler;
use crate::http_server::endpoints::tts::enqueue_infer_tts_handler::enqueue_infer_tts_handler::enqueue_infer_tts_handler;
use crate::http_server::endpoints::tts::enqueue_upload_tts_model::upload_tts_model_handler;
use crate::http_server::endpoints::tts::get_pending_tts_inference_job_count::get_pending_tts_inference_job_count_handler;
@@ -174,7 +175,7 @@ pub fn add_routes<T, B> (app: App<T>) -> App<T>
app = add_subscription_routes(app); /* /v1/subscriptions/... */
// TODO find a long term feature flag solution, since this code is likely deployed into production we don't want the route found.
let enable_voice_designer_route = false;
let enable_voice_designer_route = true;
if (enable_voice_designer_route) {
app = add_voice_designer_routes(app); /* /v1/voice_designer */
}