record entity token/type after job completes

This commit is contained in:
Brandon Thomas committed 2023-04-08 22:59:16 -07:00
1 parent b461c7e8e7
commit 8229a3cfb3
13 files changed
+265 -86

No files matched your search

@@ -6,7 +6,7 @@ use strum::EnumCount;
use strum::EnumIter;
/// Used in the `generic_inference_jobs` table in a `VARCHAR(32)` field.
/// Used in the `generic_inference_jobs` table in `VARCHAR(32)` field `inference_category`.
///
/// Our "generic inference" pipeline supports a wide variety of ML models and other media.
/// Each "category" of inference is identified by the following enum variants.
@@ -17,11 +17,9 @@ use strum::EnumIter;
#[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize)]
pub enum InferenceCategory {
#[serde(rename = "text_to_speech")]
//#[sqlx(rename = "text_to_speech")]
TextToSpeech,
#[serde(rename = "voice_conversion")]
//#[sqlx(rename = "voice_conversion")]
VoiceConversion,
}
@@ -0,0 +1,95 @@
use std::collections::BTreeSet;
#[cfg(test)]
use strum::EnumCount;
#[cfg(test)]
use strum::EnumIter;
/// Used in the `generic_inference_jobs` table in `VARCHAR(32)` field `maybe_model_type`.
///
/// Our "generic inference" pipeline supports a wide variety of ML models and other media.
/// Each inference "model type" identified by the following enum variants, though some pipelines
/// may use multiple models or no model (and may report NULL).
///
/// These types are present in the HTTP API and database columns as serialized here.
///
/// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY.
#[cfg_attr(test, derive(EnumIter, EnumCount))]
#[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize)]
pub enum InferenceModelType {
#[serde(rename = "tacotron2")]
Tacotron2,
#[serde(rename = "vits")]
Vits,
}
// TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
impl_enum_display_and_debug_using_to_str!(InferenceModelType);
impl_mysql_enum_coders!(InferenceModelType);
/// NB: Legacy API for older code.
impl InferenceModelType {
pub fn to_str(&self) -> &'static str {
match self {
Self::Tacotron2 => "tacotron2",
Self::Vits => "vits",
}
}
pub fn from_str(value: &str) -> Result<Self, String> {
match value {
"tacotron2" => Ok(Self::Tacotron2),
"vits" => Ok(Self::Vits),
_ => Err(format!("invalid value: {:?}", value)),
}
}
pub fn all_variants() -> BTreeSet<InferenceModelType> {
// NB: BTreeSet is sorted
// NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc.
BTreeSet::from([
InferenceModelType::Tacotron2,
InferenceModelType::Vits,
])
}
}
#[cfg(test)]
mod tests {
use crate::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use crate::test_helpers::assert_serialization;
#[test]
fn test_serialization() {
assert_serialization(InferenceModelType::Tacotron2, "tacotron2");
assert_serialization(InferenceModelType::Vits, "vits");
}
#[test]
fn to_str() {
assert_eq!(InferenceModelType::Tacotron2.to_str(), "tacotron2");
assert_eq!(InferenceModelType::Vits.to_str(), "vits");
}
#[test]
fn from_str() {
assert_eq!(InferenceModelType::from_str("tacotron2").unwrap(), InferenceModelType::Tacotron2);
assert_eq!(InferenceModelType::from_str("vits").unwrap(), InferenceModelType::Vits);
}
#[test]
fn all_variants() {
// Static check
let mut variants = InferenceModelType::all_variants();
assert_eq!(variants.len(), 2);
assert_eq!(variants.pop_first(), Some(InferenceModelType::Tacotron2));
assert_eq!(variants.pop_first(), Some(InferenceModelType::Vits));
assert_eq!(variants.pop_first(), None);
// Generated check
use strum::IntoEnumIterator;
assert_eq!(InferenceModelType::all_variants().len(), InferenceModelType::iter().len());
}
}
@@ -0,0 +1,94 @@
use std::collections::BTreeSet;
#[cfg(test)]
use strum::EnumCount;
#[cfg(test)]
use strum::EnumIter;
/// Used in the `generic_inference_jobs` table in `VARCHAR(32)` field `on_success_result_entity_type`.
///
/// Our "generic inference" pipeline supports a wide variety of output types.
/// Each "result type" is identified by the following enum variants.
///
/// These types are present in the HTTP API and database columns as serialized here.
///
/// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY.
#[cfg_attr(test, derive(EnumIter, EnumCount))]
#[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize)]
pub enum InferenceResultType {
#[serde(rename = "text_to_speech")]
TextToSpeech,
#[serde(rename = "voice_conversion")]
VoiceConversion,
}
// TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
impl_enum_display_and_debug_using_to_str!(InferenceResultType);
impl_mysql_enum_coders!(InferenceResultType);
/// NB: Legacy API for older code.
impl InferenceResultType {
pub fn to_str(&self) -> &'static str {
match self {
Self::TextToSpeech => "text_to_speech",
Self::VoiceConversion => "voice_conversion",
}
}
pub fn from_str(value: &str) -> Result<Self, String> {
match value {
"text_to_speech" => Ok(Self::TextToSpeech),
"voice_conversion" => Ok(Self::VoiceConversion),
_ => Err(format!("invalid value: {:?}", value)),
}
}
pub fn all_variants() -> BTreeSet<InferenceResultType> {
// NB: BTreeSet is sorted
// NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc.
BTreeSet::from([
InferenceResultType::TextToSpeech,
InferenceResultType::VoiceConversion,
])
}
}
#[cfg(test)]
mod tests {
use crate::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
use crate::test_helpers::assert_serialization;
#[test]
fn test_serialization() {
assert_serialization(InferenceResultType::TextToSpeech, "text_to_speech");
assert_serialization(InferenceResultType::VoiceConversion, "voice_conversion");
}
#[test]
fn to_str() {
assert_eq!(InferenceResultType::TextToSpeech.to_str(), "text_to_speech");
assert_eq!(InferenceResultType::VoiceConversion.to_str(), "voice_conversion");
}
#[test]
fn from_str() {
assert_eq!(InferenceResultType::from_str("text_to_speech").unwrap(), InferenceResultType::TextToSpeech);
assert_eq!(InferenceResultType::from_str("voice_conversion").unwrap(), InferenceResultType::VoiceConversion);
}
#[test]
fn all_variants() {
// Static check
let mut variants = InferenceResultType::all_variants();
assert_eq!(variants.len(), 2);
assert_eq!(variants.pop_first(), Some(InferenceResultType::TextToSpeech));
assert_eq!(variants.pop_first(), Some(InferenceResultType::VoiceConversion));
assert_eq!(variants.pop_first(), None);
// Generated check
use strum::IntoEnumIterator;
assert_eq!(InferenceResultType::all_variants().len(), InferenceResultType::iter().len());
}
}
@@ -1 +1,3 @@
pub mod inference_category;
pub mod inference_model_type;
pub mod inference_result_type;
@@ -1,12 +1,13 @@
use anyhow::anyhow;
use errors::AnyhowResult;
use crate::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
use errors::AnyhowResult;
use sqlx::MySqlPool;
pub async fn mark_generic_inference_job_successfully_done(
pool: &MySqlPool,
job: &AvailableInferenceJob,
maybe_entity_type: Option<&str>,
maybe_entity_type: Option<InferenceResultType>,
maybe_entity_token: Option<&str>,
) -> AnyhowResult<()>
{
@@ -0,0 +1,10 @@
use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
pub struct JobSuccessResult {
pub maybe_result_entity: Option<ResultEntity>,
}
pub struct ResultEntity {
pub entity_type: InferenceResultType,
pub entity_token: String,
}
@@ -1,4 +1,5 @@
pub mod job_stats;
pub mod job_success_result;
pub mod main_loop;
pub mod process_single_job;
pub mod process_single_job_error;
pub mod process_single_job_error;
@@ -6,9 +6,11 @@ use crate::job_dependencies::JobDependencies;
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use errors::AnyhowResult;
use log::{info, warn};
use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::generic_inference::job::mark_generic_inference_job_pending_and_grab_lock::mark_generic_inference_job_pending_and_grab_lock;
use mysql_queries::queries::generic_inference::job::mark_generic_inference_job_successfully_done::mark_generic_inference_job_successfully_done;
use crate::job::job_loop::job_success_result::ResultEntity;
pub async fn process_single_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result<(), ProcessSingleJobError> {
// TODO(bt, 2023-01-11): Restore an optional status logger
@@ -38,75 +40,34 @@ pub async fn process_single_job(job_dependencies: &JobDependencies, job: &Availa
// ==================== HANDLE DIFFERENT INFERENCE TYPES ==================== //
let mut entity_token : Option<String> = None;
let mut entity_type : Option<String> = None;
let mut maybe_result_entity : Option<ResultEntity>;
match job.inference_category {
let job_success_result = match job.inference_category {
InferenceCategory::TextToSpeech => {
// TODO
entity_type = Some("todo".to_string()); // TODO
entity_token = Some("todo".to_string()); // TODO
let _r = process_single_tts_job(job_dependencies, job).await?;
process_single_tts_job(job_dependencies, job).await?
}
InferenceCategory::VoiceConversion => {
// TODO
let _r = process_single_vc_job(job_dependencies, job).await?;
process_single_vc_job(job_dependencies, job).await?
}
// GenericDownloadType::HifiGan => {
// let results = process_hifigan_vocoder(
// job_state,
// job,
// &temp_dir,
// &download_filename,
// &mut redis_logger,
// ).await?;
// entity_token = results.entity_token.clone();
// entity_type = results.entity_type.clone();
// }
// GenericDownloadType::HifiGanRocketVc => {
// let results = process_hifigan_softvc_vocoder(
// job_state,
// job,
// &temp_dir,
// &download_filename,
// &mut redis_logger,
// ).await?;
// entity_token = results.entity_token.clone();
// entity_type = results.entity_type.clone();
// }
// GenericDownloadType::RocketVc => {
// let results = process_softvc_model(
// job_state,
// job,
// &temp_dir,
// &download_filename,
// &mut redis_logger,
// ).await?;
// entity_token = results.entity_token.clone();
// entity_type = results.entity_type.clone();
// }
// GenericDownloadType::Tacotron2 => {
// let results = process_tacotron_model(
// job_state,
// job,
// &temp_dir,
// &download_filename,
// &mut redis_logger,
// ).await?;
// entity_token = results.entity_token.clone();
// entity_type = results.entity_type.clone();
// }
}
};
let maybe_entity_type = job_success_result.maybe_result_entity
.as_ref()
.map(|result_entity| result_entity.entity_type);
let maybe_entity_token = job_success_result.maybe_result_entity
.as_ref()
.map(|result_entity| result_entity.entity_token.as_str());
// =====================================================
info!("Marking job complete...");
mark_generic_inference_job_successfully_done(
&job_dependencies.mysql_pool,
job,
entity_type.as_deref(),
entity_token.as_deref(),
maybe_entity_type,
maybe_entity_token,
).await
.map_err(|err| ProcessSingleJobError::Other(anyhow!("database error: {:?}", err)))?;
@@ -9,8 +9,9 @@ use errors::AnyhowResult;
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, TtsModelForInferenceRecord};
use tokens::tokens::tts_models::TtsModelToken;
use crate::job::job_loop::job_success_result::JobSuccessResult;
pub async fn process_single_tts_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result<(), ProcessSingleJobError> {
pub async fn process_single_tts_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result<JobSuccessResult, ProcessSingleJobError> {
let tts_model_token = match job.maybe_model_token.as_deref() {
None => return Err(ProcessSingleJobError::Other(anyhow!("no model token on job"))),
@@ -60,25 +61,25 @@ pub async fn process_single_tts_job(job_dependencies: &JobDependencies, job: &Av
Some(model) => model,
};
match tts_model.tts_model_type {
let job_success_result = match tts_model.tts_model_type {
TtsModelType::Tacotron2 => {
let _r = tacotron2_v2_early_fakeyou::process_job::process_job(ProcessJobArgs {
tacotron2_v2_early_fakeyou::process_job::process_job(ProcessJobArgs {
job_dependencies,
job,
tts_model: &tts_model,
raw_inference_text,
}).await?;
}).await?
}
TtsModelType::Vits => {
let _r = vits::process_job::process_job(VitsProcessJobArgs {
vits::process_job::process_job(VitsProcessJobArgs {
job_dependencies,
job,
tts_model: &tts_model,
raw_inference_text,
}).await?;
}).await?
}
}
};
Ok(())
Ok(job_success_result)
}
@@ -2,6 +2,7 @@ use anyhow::anyhow;
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 crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::tts::tacotron2_v2_early_fakeyou::seconds_to_decoder_steps::seconds_to_decoder_steps;
use crate::job::job_types::tts::tacotron2_v2_early_fakeyou::tacotron2_inference_command::{InferenceArgs, MelMultiplyFactor, Tacotron2InferenceCommand, VocoderForInferenceOption};
@@ -20,6 +21,7 @@ use std::io::Read;
use std::path::PathBuf;
use subprocess_common::docker_options::{DockerFilesystemMount, DockerGpu, DockerOptions};
use tempdir::TempDir;
use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
use tts_common::clean_symbols::clean_symbols;
use tts_common::text_pipelines::guess_pipeline::guess_text_pipeline_heuristic;
use tts_common::text_pipelines::text_pipeline_type::TextPipelineType;
@@ -36,7 +38,7 @@ pub struct ProcessJobArgs<'a> {
pub raw_inference_text: &'a str,
}
pub async fn process_job(args: ProcessJobArgs<'_>) -> Result<(), ProcessSingleJobError> {
pub async fn process_job(args: ProcessJobArgs<'_>) -> Result<JobSuccessResult, ProcessSingleJobError> {
let job = args.job;
let tts_model = args.tts_model;
let raw_inference_text = args.raw_inference_text;
@@ -390,7 +392,12 @@ pub async fn process_job(args: ProcessJobArgs<'_>) -> Result<(), ProcessSingleJo
info!("Job {:?} complete success! Downloaded, ran inference, and uploaded. Saved model record: {}, Result Token: {}",
job.id, id, &inference_result_token);
Ok(())
Ok(JobSuccessResult {
maybe_result_entity: Some(ResultEntity {
entity_type: InferenceResultType::TextToSpeech,
entity_token: inference_result_token
}),
})
}
#[derive(Deserialize, Default)]
@@ -22,9 +22,11 @@ use std::io::Read;
use std::path::{Path, PathBuf};
use subprocess_common::docker_options::{DockerFilesystemMount, DockerGpu, DockerOptions};
use tempdir::TempDir;
use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType;
use tts_common::clean_symbols::clean_symbols;
use tts_common::text_pipelines::guess_pipeline::guess_text_pipeline_heuristic;
use tts_common::text_pipelines::text_pipeline_type::TextPipelineType;
use crate::job::job_loop::job_success_result::{JobSuccessResult, ResultEntity};
/// Text starting with this will be treated as a test request.
/// This allows the request to bypass the model cache and query the latest TTS model.
@@ -38,7 +40,7 @@ pub struct VitsProcessJobArgs<'a> {
pub raw_inference_text: &'a str,
}
pub async fn process_job(args: VitsProcessJobArgs<'_>) -> Result<(), ProcessSingleJobError> {
pub async fn process_job(args: VitsProcessJobArgs<'_>) -> Result<JobSuccessResult, ProcessSingleJobError> {
let job = args.job;
let tts_model = args.tts_model;
let raw_inference_text = args.raw_inference_text;
@@ -245,7 +247,12 @@ pub async fn process_job(args: VitsProcessJobArgs<'_>) -> Result<(), ProcessSing
info!("Job {:?} complete success! Downloaded, ran inference, and uploaded. Saved model record: {}, Result Token: {}",
job.id, id, &inference_result_token);
Ok(())
Ok(JobSuccessResult {
maybe_result_entity: Some(ResultEntity {
entity_type: InferenceResultType::TextToSpeech,
entity_token: inference_result_token
}),
})
}
#[derive(Deserialize, Default)]
@@ -2,8 +2,11 @@ use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job_dependencies::JobDependencies;
use errors::AnyhowResult;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use crate::job::job_loop::job_success_result::JobSuccessResult;
pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result<(), ProcessSingleJobError> {
Ok(())
pub async fn process_single_vc_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> Result<JobSuccessResult, ProcessSingleJobError> {
Ok(JobSuccessResult {
maybe_result_entity: None,
})
}
@@ -35,6 +35,12 @@ pub struct GetInferenceJobStatusPathInfo {
token: InferenceJobToken,
}
#[derive(Serialize)]
pub struct GetInferenceJobStatusSuccessResponse {
pub success: bool,
pub state: InferenceJobStatusResponsePayload,
}
#[derive(Serialize)]
pub struct InferenceJobStatusResponsePayload {
pub job_token: InferenceJobToken,
@@ -49,7 +55,6 @@ pub struct InferenceJobStatusResponsePayload {
#[derive(Serialize)]
pub struct RequestDetailsResponse {
// TODO: Rename this inference_category (!!!) and add model_type.
pub inference_category: InferenceCategory,
pub maybe_model_type: Option<String>,
pub maybe_model_token: Option<String>,
@@ -78,13 +83,7 @@ pub struct ResultDetailsResponse {
pub entity_token: String,
/// NB: This is only for audio- or video- type results.
pub maybe_public_bucket_wav_media_path: Option<String>,
}
#[derive(Serialize)]
pub struct GetInferenceJobStatusSuccessResponse {
pub success: bool,
pub state: InferenceJobStatusResponsePayload,
pub maybe_public_bucket_media_path: Option<String>,
}
#[derive(Debug)]
@@ -180,7 +179,7 @@ pub async fn get_inference_job_status_handler(
ResultDetailsResponse {
entity_type: result_details.entity_type,
entity_token: result_details.entity_token,
maybe_public_bucket_wav_media_path: None, // TODO: This will require a new field
maybe_public_bucket_media_path: None, // TODO: This will require a new field
}
}),
created_at: record.created_at,