diff --git a/_sql/actions_and_analytics/generic_inference/find_non_model_weights_tokens_in_jobs.sql b/_sql/actions_and_analytics/generic_inference/find_non_model_weights_tokens_in_jobs.sql new file mode 100644 index 0000000000..3144d7e5ff --- /dev/null +++ b/_sql/actions_and_analytics/generic_inference/find_non_model_weights_tokens_in_jobs.sql @@ -0,0 +1,16 @@ +-- Find vcm_* tokens that are still in use +select + job_token, + model_token, + user_token +from ( + select token as job_token, + maybe_model_token as model_token, + maybe_creator_user_token as user_token + from generic_inference_jobs + order by id + desc limit 1000000 +) as x +where model_token NOT LIKE "TM:%" +and model_token NOT LIKE "weight_%" +limit 50; diff --git a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs b/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs index df2dacdd84..3812a0ab83 100644 --- a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs +++ b/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs @@ -24,6 +24,7 @@ use crate::job::job_types::mocap::process_single_mc_job::process_single_mc_job; use crate::job::job_types::render_engine_scene::process_render_engine_scene::process_single_render_engine_scene_job; use crate::job::job_types::tts::process_single_tts_job::process_single_tts_job; use crate::job::job_types::vc::process_single_vc_job::process_single_vc_job; +use crate::job::job_types::vc::rvc_v2::dispatch_rvc_v2_job::dispatch_rvc_v2_job; use crate::job::job_types::videofilter::process_single_vf_job::process_single_vf_job; use crate::job::job_types::workflow::process_single_workflow_job::process_single_workflow_job; use crate::state::job_dependencies::JobDependencies; @@ -275,6 +276,9 @@ async fn new_dispatch( InferenceJobType::GptSovits => { process_single_gpt_sovits_job(job_dependencies, job).await? }, + InferenceJobType::RvcV2 => { + dispatch_rvc_v2_job(job_dependencies, job).await? + }, _ => { return Err(ProcessSingleJobError::InvalidJob( anyhow!("invalid job type for dispatch: {:?}", job.job_type))) diff --git a/crates/service/job/inference_job/src/job/job_types/gpt_sovits/process_single_gpt_sovits_job.rs b/crates/service/job/inference_job/src/job/job_types/gpt_sovits/process_single_gpt_sovits_job.rs index fd694e292e..fc9251fe9a 100644 --- a/crates/service/job/inference_job/src/job/job_types/gpt_sovits/process_single_gpt_sovits_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/gpt_sovits/process_single_gpt_sovits_job.rs @@ -27,7 +27,7 @@ pub async fn process_single_gpt_sovits_job( } } _ => { - Err(ProcessSingleJobError::Other(anyhow!("tts model type not set"))) + Err(ProcessSingleJobError::Other(anyhow!("job type not set"))) } } } \ No newline at end of file diff --git a/crates/service/job/inference_job/src/job/job_types/vc/process_single_vc_job.rs b/crates/service/job/inference_job/src/job/job_types/vc/process_single_vc_job.rs index b1eecd0bd3..296ac98a97 100644 --- a/crates/service/job/inference_job/src/job/job_types/vc/process_single_vc_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/vc/process_single_vc_job.rs @@ -1,5 +1,5 @@ use anyhow::anyhow; -use log::{error, info}; +use log::{error, info, warn}; use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType; use migration::voice_conversion::query_vc_model_for_migration::{query_vc_model_for_migration, VcModelType}; @@ -13,11 +13,15 @@ use crate::job::job_loop::job_success_result::JobSuccessResult; use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; use crate::job::job_types::vc::{rvc_v2, so_vits_svc}; use crate::job::job_types::vc::media_for_inference::MediaForInference; -use crate::job::job_types::vc::rvc_v2::process_job::RvcV2ProcessJobArgs; +use crate::job::job_types::vc::rvc_v2::process_rvc_job::RvcV2ProcessJobArgs; use crate::job::job_types::vc::so_vits_svc::process_job::SoVitsSvcProcessJobArgs; use crate::state::job_dependencies::JobDependencies; -pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result { +pub async fn process_single_vc_job( + job_dependencies: &JobDependencies, + job: &AvailableInferenceJob +) -> Result { + let model_token = match job.maybe_model_token.as_deref() { Some(token) => token, None => return Err(ProcessSingleJobError::InvalidJob(anyhow!("No model token for job: {:?}", job.inference_job_token))), @@ -113,7 +117,8 @@ pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &Ava let job_success_result = match model.get_model_type() { VcModelType::RvcV2 => { - rvc_v2::process_job::process_job(RvcV2ProcessJobArgs { + warn!("OLD/LEGACY RVC code path. We should instead dispatch on the basis of job_type. See `dispatch_rvc_v2_job` and migrate to that code path."); + rvc_v2::process_rvc_job::process_rvc_job(RvcV2ProcessJobArgs { job_dependencies, job, vc_model: &model, @@ -134,5 +139,3 @@ pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &Ava Ok(job_success_result) } - - diff --git a/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/dispatch_rvc_v2_job.rs b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/dispatch_rvc_v2_job.rs new file mode 100644 index 0000000000..83c7237adc --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/dispatch_rvc_v2_job.rs @@ -0,0 +1,38 @@ +use anyhow::anyhow; +use log::info; +use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType; +use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; + +use crate::job::job_loop::job_success_result::JobSuccessResult; +use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; +use crate::job::job_types::vc::rvc_v2::process_rvc_job_old_entrypoint_adapter::process_rvc_job_old_entrypoint_adapter; +use crate::state::job_dependencies::JobDependencies; + +pub async fn dispatch_rvc_v2_job( + job_dependencies: &JobDependencies, + job: &AvailableInferenceJob +) -> Result { + + info!("New dispatch route(2024-09-04): dispatch_rvc_v2_job instead of process_single_vc_job"); + + return match job.job_type { + InferenceJobType::RvcV2 => { + match job.maybe_download_url { + Some(_) => { + // TODO(bt,2024-09-04): Support RVC uploads + Err(ProcessSingleJobError::Other(anyhow!("RVC uploads not yet supported!"))) + } + None => { + let job_success_result = process_rvc_job_old_entrypoint_adapter( + job_dependencies, + job + ).await?; + Ok(job_success_result) + } + } + } + _ => { + Err(ProcessSingleJobError::Other(anyhow!("job type not set (new rvc code path)"))) + } + } +} diff --git a/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/mod.rs b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/mod.rs index ba3ab43f2c..a42b7e24bf 100644 --- a/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/mod.rs +++ b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/mod.rs @@ -1,5 +1,7 @@ +pub mod dispatch_rvc_v2_job; pub mod model_downloaders; pub mod pretrained_hubert_model; -pub mod process_job; +pub mod process_rvc_job; +pub mod process_rvc_job_old_entrypoint_adapter; pub mod rvc_v2_dependencies; pub mod rvc_v2_inference_command; \ No newline at end of file diff --git a/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_job.rs b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_rvc_job.rs similarity index 99% rename from crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_job.rs rename to crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_rvc_job.rs index 14f5074889..3a57d27ddb 100644 --- a/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_rvc_job.rs @@ -42,7 +42,7 @@ pub struct RvcV2ProcessJobArgs<'a> { pub inference_media: &'a MediaForInference, } -pub async fn process_job(args: RvcV2ProcessJobArgs<'_>) -> Result { +pub async fn process_rvc_job(args: RvcV2ProcessJobArgs<'_>) -> Result { let job = args.job; let vc_model = args.vc_model; diff --git a/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_rvc_job_old_entrypoint_adapter.rs b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_rvc_job_old_entrypoint_adapter.rs new file mode 100644 index 0000000000..e0ab5cd0a7 --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/vc/rvc_v2/process_rvc_job_old_entrypoint_adapter.rs @@ -0,0 +1,118 @@ +use anyhow::anyhow; +use log::{error, info}; + +use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType; +use migration::voice_conversion::query_vc_model_for_migration::{query_vc_model_for_migration, VcModelType}; +use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; +use mysql_queries::queries::media_files::get::get_media_file_for_inference::get_media_file_for_inference; +use mysql_queries::queries::media_uploads::get_media_upload_for_inference::get_media_upload_for_inference; +use tokens::tokens::media_files::MediaFileToken; +use tokens::tokens::media_uploads::MediaUploadToken; + +use crate::job::job_loop::job_success_result::JobSuccessResult; +use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; +use crate::job::job_types::vc::{rvc_v2, so_vits_svc}; +use crate::job::job_types::vc::media_for_inference::MediaForInference; +use crate::job::job_types::vc::rvc_v2::process_rvc_job::RvcV2ProcessJobArgs; +use crate::job::job_types::vc::so_vits_svc::process_job::SoVitsSvcProcessJobArgs; +use crate::state::job_dependencies::JobDependencies; + +/// This is meant to take over from the `process_single_vc_job` function for a short bit +pub async fn process_rvc_job_old_entrypoint_adapter( + job_dependencies: &JobDependencies, + job: &AvailableInferenceJob +) -> Result { + + let model_token = match job.maybe_model_token.as_deref() { + Some(token) => token, + None => return Err(ProcessSingleJobError::InvalidJob(anyhow!("No model token for job: {:?}", job.inference_job_token))), + }; + + // TODO: Interrogate a db result cache + filesystem model file cache + + let maybe_model = query_vc_model_for_migration(model_token, &job_dependencies.db.mysql_pool) + .await + .map_err(|err| { + error!("error querying vc model {model_token} : {:?}", err); + ProcessSingleJobError::Other(anyhow!("error querying vc model {model_token} : {:?}", err)) + })?; + + let model = match maybe_model { + None => return Err(ProcessSingleJobError::Other(anyhow!("model weights not found: {:?}", model_token))), + Some(model) => model, + }; + + let media_token = job.maybe_input_source_token + .as_deref() + .ok_or_else(|| ProcessSingleJobError::Other(anyhow!( + "no associated media token for vc job: {:?}", job.inference_job_token)))?; + + let token_type = job.maybe_input_source_token_type + .ok_or_else(|| ProcessSingleJobError::Other(anyhow!( + "no associated media token type for vc job: {:?}", job.inference_job_token)))?; + + let inference_media = match token_type { + InferenceInputSourceTokenType::MediaFile => { + // media_files case + let media_file_token = MediaFileToken::new_from_str(media_token); + let maybe_media_file = get_media_file_for_inference(&media_file_token, &job_dependencies.db.mysql_pool).await; + + let media_file = match maybe_media_file { + Ok(Some(media_file)) => media_file, + Ok(None) => { + error!("no media file record found for token: {:?}", media_token); + return Err(ProcessSingleJobError::Other( + anyhow!("no media file record found for token: {:?}", media_token))); + } + Err(err) => { + error!("error fetching media file record from db: {:?}", err); + return Err(ProcessSingleJobError::Other(err)); + } + }; + + MediaForInference::MediaFile(media_file) + } + InferenceInputSourceTokenType::MediaUpload => { + // media_uploads case + let media_upload_token = MediaUploadToken::new_from_str(media_token); + let maybe_media_upload_result = + get_media_upload_for_inference(&media_upload_token, &job_dependencies.db.mysql_pool).await; + + let media_upload = match maybe_media_upload_result { + Ok(Some(media_upload)) => media_upload, + Ok(None) => { + error!("no media upload record found for token: {:?}", media_token); + return Err(ProcessSingleJobError::Other( + anyhow!("no media upload record found for token: {:?}", media_token))); + } + Err(err) => { + error!("error fetching media upload record from db: {:?}", err); + return Err(ProcessSingleJobError::Other(err)); + } + }; + + MediaForInference::LegacyMediaUpload(media_upload) + } + }; + + info!("Source media upload file size (bytes): {}", inference_media.file_size_bytes()); + info!("Source media upload duration (millis): {:?}", inference_media.maybe_duration_millis()); + info!("Source media upload duration (seconds): {:?}", (inference_media.maybe_duration_millis() + .map(|d| d as f32 / 1000.0))); + + let job_success_result = match model.get_model_type() { + VcModelType::RvcV2 => { + rvc_v2::process_rvc_job::process_rvc_job(RvcV2ProcessJobArgs { + job_dependencies, + job, + vc_model: &model, + inference_media: &inference_media, + }).await? + } + _ => return Err(ProcessSingleJobError::InvalidJob(anyhow!("only RVC models should be dispatched this way"))), + }; + + Ok(job_success_result) +} + + diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/enqueue_voice_conversion_inference_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/enqueue_voice_conversion_inference_handler.rs index 10bd39c1f0..66620c9fa8 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/enqueue_voice_conversion_inference_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/voice_conversion/enqueue_voice_conversion_inference_handler.rs @@ -281,25 +281,6 @@ pub async fn enqueue_voice_conversion_inference_handler( "input token is not a media_upload or media_file token".to_string())); }; - // NB: This is better, but we'd need to change the API and migrate callers. - // let (input_token, input_token_type) = - // match (&request.source_media_upload_token, &request.source_media_file_token) { - // (Some(media_upload), None) => { - // (media_upload.to_string(), InferenceInputSourceTokenType::MediaUpload) - // }, - // (None, Some(media_file)) => { - // (media_file.to_string(), InferenceInputSourceTokenType::MediaFile) - // }, - // (Some(_media_upload), Some(_media_file)) => { - // return Err(EnqueueVoiceConversionInferenceError::BadInput( - // "both source_media_upload_token and source_media_file_token provided".to_string())); - // }, - // (None, None) => { - // return Err(EnqueueVoiceConversionInferenceError::BadInput( - // "neither source_media_upload_token nor source_media_file_token provided".to_string())); - // }, - // }; - // ==================== CHECK AND ENQUEUE VOICE CONVERSION ==================== // let mut redis = server_state.redis_pool