From 7bea2ed7240688e24bc064376ab0aa1779aec78e Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Thu, 25 Jan 2024 22:48:34 -0500 Subject: [PATCH] Inference job stub for format conversion (fbx2gltf) --- .../generic_inference_jobs.sql | 6 + .../inference_category.rs | 30 +- .../generic_inference_args.rs | 6 + .../web/get_inference_job_status.rs | 2 + .../get_media_file_for_inference.rs | 98 ++++++ .../src/queries/media_files/mod.rs | 1 + .../src/job/job_loop/process_single_job.rs | 8 +- .../format_conversion/fbx_to_gltf/mod.rs | 1 + .../fbx_to_gltf/process_job.rs | 317 ++++++++++++++++++ .../job/job_types/format_conversion/mod.rs | 2 + .../process_single_format_conversion_job.rs | 60 ++++ .../inference_job/src/job/job_types/mod.rs | 1 + 12 files changed, 520 insertions(+), 12 deletions(-) create mode 100644 crates/lib/mysql_queries/src/queries/media_files/get_media_file_for_inference.rs create mode 100644 crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/mod.rs create mode 100644 crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/process_job.rs create mode 100644 crates/service/job/inference_job/src/job/job_types/format_conversion/mod.rs create mode 100644 crates/service/job/inference_job/src/job/job_types/format_conversion/process_single_format_conversion_job.rs diff --git a/_sql/migrations_squashed/generic_inference_jobs.sql b/_sql/migrations_squashed/generic_inference_jobs.sql index 61535e1e36..6ee7f74407 100644 --- a/_sql/migrations_squashed/generic_inference_jobs.sql +++ b/_sql/migrations_squashed/generic_inference_jobs.sql @@ -17,12 +17,18 @@ CREATE TABLE generic_inference_jobs ( -- ========== INFERENCE DETAILS ========== + -- NB: Note that this is probably coming to mean "job type". It's grown to indicate the + -- exact type of job to run (though that's still also somewhat incorrectly spread over + -- maybe_model_type). + -- -- Broad category of inference -- Examples (may not be up to date): -- * text_to_speech -- * voice_conversion inference_category VARCHAR(32) NOT NULL, + -- NB: Note that this is probably no longer useful (so long as we migrate away from it) + -- -- Potential part of the composite foreign key to the primary model being used, if any. -- This will normally live in `maybe_inference_args`, but in this case, it's useful for -- running easy database analytical queries. diff --git a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_category.rs b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_category.rs index 26652eca36..c310ea764d 100644 --- a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_category.rs +++ b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_category.rs @@ -38,9 +38,9 @@ pub enum InferenceCategory { #[serde(rename = "workflow")] Workflow, - /// Eg. A job that turns "FBX" animation files into "GLTF" animation files. - #[serde(rename = "format_conversion")] - FormatConversion, + /// A job that turns "FBX" game engine files into "GLTF" files (Bevy-compatible). + #[serde(rename = "convert_fbx_gltf")] + ConvertFbxToGltf, } // TODO(bt, 2022-12-21): This desperately needs MySQL integration tests! @@ -58,7 +58,7 @@ impl InferenceCategory { Self::ImageGeneration => "image_generation", Self::Mocap => "mocap", Self::Workflow => "workflow", - Self::FormatConversion => "format_conversion", + Self::ConvertFbxToGltf => "convert_fbx_gltf", } } @@ -71,7 +71,7 @@ impl InferenceCategory { "image_generation" => Ok(Self::ImageGeneration), "mocap" => Ok(Self::Mocap), "workflow" => Ok(Self::Workflow), - "format_conversion" => Ok(Self::FormatConversion), + "convert_fbx_gltf" => Ok(Self::ConvertFbxToGltf), _ => Err(format!("invalid value: {:?}", value)), } } @@ -87,7 +87,7 @@ impl InferenceCategory { Self::ImageGeneration, Self::Mocap, Self::Workflow, - Self::FormatConversion, + Self::ConvertFbxToGltf, ]) } } @@ -109,7 +109,7 @@ mod tests { assert_serialization(InferenceCategory::ImageGeneration, "image_generation"); assert_serialization(InferenceCategory::Mocap, "mocap"); assert_serialization(InferenceCategory::Workflow, "workflow"); - assert_serialization(InferenceCategory::FormatConversion, "format_conversion"); + assert_serialization(InferenceCategory::ConvertFbxToGltf, "convert_fbx_gltf"); } #[test] @@ -121,7 +121,7 @@ mod tests { assert_eq!(InferenceCategory::ImageGeneration.to_str(), "image_generation"); assert_eq!(InferenceCategory::Mocap.to_str(), "mocap"); assert_eq!(InferenceCategory::Workflow.to_str(), "workflow"); - assert_eq!(InferenceCategory::FormatConversion.to_str(), "format_conversion"); + assert_eq!(InferenceCategory::ConvertFbxToGltf.to_str(), "convert_fbx_gltf"); } #[test] @@ -133,7 +133,7 @@ mod tests { assert_eq!(InferenceCategory::from_str("image_generation").unwrap(), InferenceCategory::ImageGeneration); assert_eq!(InferenceCategory::from_str("mocap").unwrap(), InferenceCategory::Mocap); assert_eq!(InferenceCategory::from_str("workflow").unwrap(), InferenceCategory::Workflow); - assert_eq!(InferenceCategory::from_str("format_conversion").unwrap(), InferenceCategory::FormatConversion); + assert_eq!(InferenceCategory::from_str("convert_fbx_gltf").unwrap(), InferenceCategory::ConvertFbxToGltf); } #[test] @@ -148,7 +148,7 @@ mod tests { assert_eq!(variants.pop_first(), Some(InferenceCategory::ImageGeneration)); assert_eq!(variants.pop_first(), Some(InferenceCategory::Mocap)); assert_eq!(variants.pop_first(), Some(InferenceCategory::Workflow)); - assert_eq!(variants.pop_first(), Some(InferenceCategory::FormatConversion)); + assert_eq!(variants.pop_first(), Some(InferenceCategory::ConvertFbxToGltf)); assert_eq!(variants.pop_first(), None); // Generated check @@ -174,5 +174,15 @@ mod tests { assert_eq!(variant, InferenceCategory::from_str(&format!("{:?}", variant)).unwrap()); } } + + #[test] + fn serialized_length_ok_for_database() { + const MAX_LENGTH : usize = 32; + for variant in InferenceCategory::all_variants() { + let serialized = variant.to_str(); + assert!(serialized.len() > 0, "variant {:?} is too short", variant); + assert!(serialized.len() <= MAX_LENGTH, "variant {:?} is too long", variant); + } + } } } diff --git a/crates/lib/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs b/crates/lib/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs index 755a952756..b23138e940 100644 --- a/crates/lib/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs +++ b/crates/lib/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs @@ -54,6 +54,10 @@ pub enum InferenceCategoryAbbreviated { #[serde(rename = "wf")] // NB: DO NOT CHANGE. It could break live jobs. Renamed to be fewer bytes. #[serde(alias = "workflow")] Workflow, + + #[serde(rename = "fg")] // NB: DO NOT CHANGE. It could break live jobs. Renamed to be fewer bytes. + #[serde(alias = "convert_fbx_gltf")] + ConvertFbxToGltf, } #[derive(Serialize, Deserialize, Clone, Copy, Debug)] @@ -138,6 +142,7 @@ impl InferenceCategoryAbbreviated { InferenceCategory::ImageGeneration => Self::ImageGeneration, InferenceCategory::Mocap => Self::Mocap, InferenceCategory::Workflow => Self::Workflow, + InferenceCategory::ConvertFbxToGltf => Self::ConvertFbxToGltf, } } @@ -150,6 +155,7 @@ impl InferenceCategoryAbbreviated { Self::ImageGeneration =>InferenceCategory::ImageGeneration, Self::Mocap => InferenceCategory::Mocap, Self::Workflow => InferenceCategory::Workflow, + Self::ConvertFbxToGltf => InferenceCategory::ConvertFbxToGltf, } } } diff --git a/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs b/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs index e3c07c3cec..a5cb2139b1 100644 --- a/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs +++ b/crates/lib/mysql_queries/src/queries/generic_inference/web/get_inference_job_status.rs @@ -182,6 +182,7 @@ fn raw_record_to_public_result(record: RawGenericInferenceJobStatus) -> GenericI InferenceCategory::ImageGeneration => Some("Image Generation"), InferenceCategory::Mocap => Some("Mocap"), InferenceCategory::Workflow => Some("Workflow"), + InferenceCategory::ConvertFbxToGltf => Some("FBX to GLTF"), }; // NB: A bit of a hack. We store TTS results with a full path. @@ -194,6 +195,7 @@ fn raw_record_to_public_result(record: RawGenericInferenceJobStatus) -> GenericI InferenceCategory::ImageGeneration => (true, record.maybe_media_file_public_bucket_directory_hash.as_deref()), InferenceCategory::Mocap => (true, record.maybe_media_file_public_bucket_directory_hash.as_deref()), InferenceCategory::Workflow => (true, record.maybe_media_file_public_bucket_directory_hash.as_deref()), + InferenceCategory::ConvertFbxToGltf => (true, record.maybe_media_file_public_bucket_directory_hash.as_deref()), }; // NB: We've moved voice conversion out of their own table and into media_files diff --git a/crates/lib/mysql_queries/src/queries/media_files/get_media_file_for_inference.rs b/crates/lib/mysql_queries/src/queries/media_files/get_media_file_for_inference.rs new file mode 100644 index 0000000000..54742d2932 --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/media_files/get_media_file_for_inference.rs @@ -0,0 +1,98 @@ +// NB: Incrementally getting rid of build warnings... +#![forbid(unused_imports)] +#![forbid(unused_mut)] +#![forbid(unused_variables)] + +use anyhow::anyhow; +use sqlx::MySqlPool; + +use enums::by_table::media_files::media_file_type::MediaFileType; +use enums::common::visibility::Visibility; +use errors::AnyhowResult; +use tokens::tokens::media_files::MediaFileToken; +use tokens::tokens::users::UserToken; + +#[derive(Serialize, Debug)] +pub struct MediaFileForInference { + pub token: MediaFileToken, + + pub media_type: MediaFileType, + + pub maybe_creator_user_token: Option, + pub creator_set_visibility: Visibility, + + pub public_bucket_directory_hash: String, + pub maybe_public_bucket_prefix: Option, + pub maybe_public_bucket_extension: Option, +} + +#[derive(Serialize)] +pub struct MediaFileRaw { + pub token: MediaFileToken, + + pub media_type: MediaFileType, + + pub maybe_creator_user_token: Option, + pub creator_set_visibility: Visibility, + + pub public_bucket_directory_hash: String, + pub maybe_public_bucket_prefix: Option, + pub maybe_public_bucket_extension: Option, +} + +pub async fn get_media_file_for_inference( + media_file_token: &MediaFileToken, + mysql_pool: &MySqlPool +) -> AnyhowResult> { + + let record = select_record(media_file_token, mysql_pool).await; + + let record = match record { + Ok(record) => record, + Err(ref err) => { + return match err { + sqlx::Error::RowNotFound => Ok(None), + _ => Err(anyhow!("database error: {:?}", err)), + } + } + }; + + Ok(Some(MediaFileForInference { + token: record.token, + media_type: record.media_type, + maybe_creator_user_token: record.maybe_creator_user_token, + creator_set_visibility: record.creator_set_visibility, + public_bucket_directory_hash: record.public_bucket_directory_hash, + maybe_public_bucket_prefix: record.maybe_public_bucket_prefix, + maybe_public_bucket_extension: record.maybe_public_bucket_extension, + })) +} + +async fn select_record( + media_file_token: &MediaFileToken, + mysql_pool: &MySqlPool +) -> Result { + sqlx::query_as!( + MediaFileRaw, + r#" +SELECT + m.token as `token: tokens::tokens::media_files::MediaFileToken`, + + m.media_type as `media_type: enums::by_table::media_files::media_file_type::MediaFileType`, + + m.maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`, + m.creator_set_visibility as `creator_set_visibility: enums::common::visibility::Visibility`, + + m.public_bucket_directory_hash, + m.maybe_public_bucket_prefix, + m.maybe_public_bucket_extension + +FROM media_files AS m +WHERE + m.token = ? + "#, + media_file_token + ) + .fetch_one(mysql_pool) + .await +} diff --git a/crates/lib/mysql_queries/src/queries/media_files/mod.rs b/crates/lib/mysql_queries/src/queries/media_files/mod.rs index 0215d99857..9acb3e90fc 100644 --- a/crates/lib/mysql_queries/src/queries/media_files/mod.rs +++ b/crates/lib/mysql_queries/src/queries/media_files/mod.rs @@ -1,5 +1,6 @@ pub mod create; pub mod delete_media_file; pub mod get_media_file; +pub mod get_media_file_for_inference; pub mod list; pub mod update_media_file; 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 0f84ea6100..ad61df3446 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 @@ -13,12 +13,13 @@ use redis_common::redis_keys::RedisKeys; use crate::job::job_loop::determine_dependency_status::determine_dependency_status; use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; use crate::job::job_loop::process_single_job_success_case::ProcessSingleJobSuccessCase; +use crate::job::job_types::format_conversion::process_single_format_conversion_job::{FormatConversionType, process_single_format_conversion_job}; +use crate::job::job_types::image_generation::process_single_ig_job::process_single_ig_job; use crate::job::job_types::lipsync::process_single_lipsync_job::process_single_lipsync_job; +use crate::job::job_types::mocap::process_single_mc_job::process_single_mc_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::videofilter::process_single_vf_job::process_single_vf_job; -use crate::job::job_types::image_generation::process_single_ig_job::process_single_ig_job; -use crate::job::job_types::mocap::process_single_mc_job::process_single_mc_job; use crate::job::job_types::workflow::process_single_wf_job::process_single_wf_job; use crate::job_dependencies::JobDependencies; @@ -198,6 +199,9 @@ async fn do_process_single_job( InferenceCategory::Workflow => { process_single_wf_job(job_dependencies, job).await? } + InferenceCategory::ConvertFbxToGltf => { + process_single_format_conversion_job(job_dependencies, job, FormatConversionType::FbxToGltf).await? + } }; let maybe_entity_type = job_success_result.maybe_result_entity diff --git a/crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/mod.rs b/crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/mod.rs new file mode 100644 index 0000000000..a5831b03dd --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/mod.rs @@ -0,0 +1 @@ +pub mod process_job; \ No newline at end of file diff --git a/crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/process_job.rs b/crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/process_job.rs new file mode 100644 index 0000000000..80d9052051 --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/format_conversion/fbx_to_gltf/process_job.rs @@ -0,0 +1,317 @@ +use std::time::Instant; + +use anyhow::anyhow; +use log::info; + +use buckets::public::media_uploads::bucket_file_path::MediaUploadOriginalFilePath; +use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; +use mysql_queries::queries::media_files::get_media_file_for_inference::MediaFileForInference; + +use crate::job::job_loop::job_success_result::JobSuccessResult; +use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; +use crate::job_dependencies::JobDependencies; +use crate::util::maybe_download_file_from_bucket::{maybe_download_file_from_bucket, MaybeDownloadArgs}; + +const BUCKET_FILE_EXTENSION : &str = ".gltf"; + +pub struct FbxToGltfJobArgs<'a> { + pub job_dependencies: &'a JobDependencies, + pub job: &'a AvailableInferenceJob, + pub media_file: &'a MediaFileForInference, +} + +pub async fn process_job(args: FbxToGltfJobArgs<'_>) -> Result { + let job = args.job; + let media_file = args.media_file; + + let mut job_progress_reporter = args.job_dependencies + .clients + .job_progress_reporter + .new_generic_inference(job.inference_job_token.as_str()) + .map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?; + + // ==================== TEMP DIR ==================== // + + let work_temp_dir = format!("temp_file_convert_{}", 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))?; + + // ==================== DOWNLOAD MEDIA FILE ==================== // + + info!("Download media for conversion..."); + + let original_media_upload_fs_path = { + let original_media_upload_fs_path = work_temp_dir.path().join("original.fbx"); + + let media_upload_bucket_path = + MediaUploadOriginalFilePath::from_object_hash(&media_file.public_bucket_directory_hash); + + let bucket_object_path = media_upload_bucket_path.to_full_object_pathbuf(); + + info!("Downloading media to bucket path: {:?}", &bucket_object_path); + + maybe_download_file_from_bucket(MaybeDownloadArgs { + name_or_description_of_file: "media upload (original file)", + final_filesystem_file_path: &original_media_upload_fs_path, + bucket_object_path: &bucket_object_path, + bucket_client: &args.job_dependencies.buckets.public_bucket_client, + job_progress_reporter: &mut job_progress_reporter, + job_progress_update_description: "downloading", + job_id: job.id.0, + scoped_tempdir_creator: &args.job_dependencies.fs.scoped_temp_dir_creator_for_work, + maybe_existing_file_minimum_size_required: None, + }).await?; + + original_media_upload_fs_path + }; + + // TODO: No duration. Replace this. + let inference_duration = Instant::now().duration_since(Instant::now()); + + /* // TODO + + // ==================== SETUP FOR CONVERSION ==================== // + + job_progress_reporter.log_status("running conversion") + .map_err(|e| ProcessSingleJobError::Other(e))?; + + //let config_path = PathBuf::from("/models/voice_conversion/so-vits-svc/example_config.json"); // TODO: This could be variable. + let input_wav_path = original_media_upload_fs_path; + + let output_audio_fs_path = work_temp_dir.path().join("output.wav"); + + let stderr_output_file = work_temp_dir.path().join("stderr.txt"); + + info!("Running VC inference..."); + + info!("Expected output audio filename: {:?}", &output_audio_fs_path); + + // TODO: Limit output length for premium. + + let maybe_args = job.maybe_inference_args + .as_ref() + .map(|args| args.args.as_ref()) + .flatten(); + + // If not specified by the user, turn off auto prediction. It sounds awful. + let auto_predict_f0 = maybe_args + .map(|args| match args { + PolymorphicInferenceArgs::Tts { .. } => None, + PolymorphicInferenceArgs::La(_) => None, + PolymorphicInferenceArgs::Vc { auto_predict_f0, .. } => *auto_predict_f0, + PolymorphicInferenceArgs::Rr(_) => None, + PolymorphicInferenceArgs::Ig(_) => None, + PolymorphicInferenceArgs::Mc(_) => None, + PolymorphicInferenceArgs::Cu(_) => None, + }) + .flatten() + .unwrap_or(false); + + let maybe_transpose = maybe_args + .map(|args| match args { + PolymorphicInferenceArgs::Tts { .. } => None, + PolymorphicInferenceArgs::La(_) => None, + PolymorphicInferenceArgs::Vc { transpose, .. } => *transpose, + PolymorphicInferenceArgs::Rr(_) => None, + PolymorphicInferenceArgs::Ig(_) => None, + PolymorphicInferenceArgs::Mc(_) => None, + PolymorphicInferenceArgs::Cu(_) => None, + }) + .flatten(); + + // ==================== RUN INFERENCE SCRIPT ==================== // + + let inference_start_time = Instant::now(); + + let command_exit_status = model_dependencies + .inference_command + .execute_inference(InferenceArgs { + model_path: &so_vits_svc_fs_path, + input_path: &input_wav_path, + output_path: &output_audio_fs_path, + stderr_output_file: &stderr_output_file, + maybe_config_path: None, + device: Device::Cuda, + auto_predict_f0, + maybe_transpose, + }); + + let inference_duration = Instant::now().duration_since(inference_start_time); + + info!("Inference took duration to complete: {:?}", &inference_duration); + + if !command_exit_status.is_success() { + error!("Inference failed: {:?}", command_exit_status); + + let error = ProcessSingleJobError::Other(anyhow!("CommandExitStatus: {:?}", command_exit_status)); + + if let Ok(contents) = read_to_string(&stderr_output_file) { + warn!("Captured stderr output: {}", contents); + + //match categorize_error(&contents) { + // Some(ProcessSingleJobError::FaceDetectionFailure) => { + // warn!("Face not detected in source image"); + // error = ProcessSingleJobError::FaceDetectionFailure; + // } + // _ => {} + //} + } + + safe_delete_temp_file(&input_wav_path); + safe_delete_temp_file(&output_audio_fs_path); + safe_delete_temp_directory(&work_temp_dir); + + return Err(error); + } + + // ==================== CHECK ALL FILES EXIST AND GET METADATA ==================== // + + info!("Checking that output files exist..."); + + check_file_exists(&output_audio_fs_path).map_err(|e| ProcessSingleJobError::Other(e))?; + + info!("Interrogating result file properties..."); + + let file_size_bytes = file_size(&output_audio_fs_path) + .map_err(|err| ProcessSingleJobError::Other(err))?; + + let maybe_mimetype = get_mimetype_for_file(&output_audio_fs_path) + .map_err(|err| ProcessSingleJobError::from_io_error(err))? + .map(|mime| mime.to_string()); + + let audio_info = decode_basic_audio_file_info(&output_audio_fs_path, maybe_mimetype.as_deref(), None) + .map_err(|err| ProcessSingleJobError::Other(err))?; + + if audio_info.required_full_decode { + warn!("Required a full decode of the output data to get duration! That's inefficient!"); + } + + info!("Calculating sha256..."); + + let file_checksum = sha256_hash_file(&output_audio_fs_path) + .map_err(|err| { + ProcessSingleJobError::Other(anyhow!("Error hashing file: {:?}", err)) + })?; + + let file_metadata = FileMetadata { + duration_millis: audio_info.duration_millis, + mimetype: maybe_mimetype.clone(), + file_size_bytes, + }; + + // ==================== UPLOAD AUDIO TO BUCKET ==================== // + + job_progress_reporter.log_status("uploading result") + .map_err(|e| ProcessSingleJobError::Other(e))?; + +// let result_bucket_location = VoiceConversionResultOriginalFilePath::generate_new(); +// +// let result_bucket_object_pathbuf = result_bucket_location.to_full_object_pathbuf(); +// +// info!("Audio destination bucket path: {:?}", &result_bucket_object_pathbuf); + + let result_bucket_location = MediaFileBucketPath::generate_new( + Some(BUCKET_FILE_PREFIX), + Some(BUCKET_FILE_EXTENSION)); + + let result_bucket_object_pathbuf = result_bucket_location.to_full_object_pathbuf(); + + info!("Audio destination bucket path: {:?}", &result_bucket_object_pathbuf); + + info!("Uploading audio media file..."); + + args.job_dependencies.buckets.public_bucket_client.upload_filename_with_content_type( + &result_bucket_object_pathbuf, + &output_audio_fs_path, + "audio/wav") + .await + .map_err(|e| ProcessSingleJobError::Other(e))?; + + safe_delete_temp_file(&output_audio_fs_path); + + // ==================== DELETE DOWNLOADED FILE ==================== // + + // NB: We should be using a tempdir, but to make absolutely certain we don't overflow the disk... + safe_delete_temp_directory(&work_temp_dir); + + // ==================== SAVE RECORDS ==================== // + + info!("Saving vc inference record..."); + +// let (inference_result_token, id) = insert_voice_conversion_result(InsertArgs { +// pool: &args.job_dependencies.mysql_pool, +// job: &job, +// public_bucket_hash: result_bucket_location.get_object_hash(), +// file_size_bytes: file_metadata.file_size_bytes, +// duration_millis: file_metadata.duration_millis.unwrap_or(0), +// is_on_prem: args.job_dependencies.container.is_on_prem, +// worker_hostname: &args.job_dependencies.container.hostname, +// worker_cluster: &args.job_dependencies.container.cluster_name, +// is_debug_worker: args.job_dependencies.worker_details.is_debug_worker +// }) +// .await +// .map_err(|e| ProcessSingleJobError::Other(e))?; + + let (inference_result_token, id) = insert_media_file_from_voice_conversion(InsertMediaFileArgs { + pool: &args.job_dependencies.db.mysql_pool, + job: &job, + voice_conversion_type: VoiceConversionModelType::SoVitsSvc, + maybe_mime_type: maybe_mimetype.as_deref(), + file_size_bytes: file_metadata.file_size_bytes, + duration_millis: file_metadata.duration_millis.unwrap_or(0), + sha256_checksum: &file_checksum, + public_bucket_directory_hash: result_bucket_location.get_object_hash(), + maybe_public_bucket_prefix: Some(BUCKET_FILE_PREFIX), + maybe_public_bucket_extension: Some(BUCKET_FILE_EXTENSION), + is_on_prem: args.job_dependencies.job.info.container.is_on_prem, + worker_hostname: &args.job_dependencies.job.info.container.hostname, + worker_cluster: &args.job_dependencies.job.info.container.cluster_name, + }) + .await + .map_err(|e| ProcessSingleJobError::Other(e))?; + + info!("VC Done."); + + // TODO: Update upstream to be strongly typed + let maybe_user_token = job.maybe_creator_user_token.as_deref() + .map(|token| UserToken::new_from_str(token)); + + args.job_dependencies.clients.firehose_publisher.vc_inference_finished( + maybe_user_token.as_ref(), + &job.inference_job_token, + inference_result_token.as_str()) + .await + .map_err(|e| { + error!("error publishing event: {:?}", e); + ProcessSingleJobError::Other(anyhow!("error publishing event")) + })?; + + job_progress_reporter.log_status("done") + .map_err(|e| ProcessSingleJobError::Other(e))?; + + info!("Job {:?} complete success! Downloaded, ran inference, and uploaded. Saved model record: {}, Result Token: {}", + job.id, id, &inference_result_token); + + */ + + Ok(JobSuccessResult { + //maybe_result_entity: Some(ResultEntity { + // entity_type: InferenceResultType::MediaFile, + // entity_token: inference_result_token.to_string(), + //}), + maybe_result_entity: None, + inference_duration, + }) +} + +#[derive(Deserialize, Default)] +struct FileMetadata { + pub duration_millis: Option, + pub mimetype: Option, + pub file_size_bytes: u64, +} diff --git a/crates/service/job/inference_job/src/job/job_types/format_conversion/mod.rs b/crates/service/job/inference_job/src/job/job_types/format_conversion/mod.rs new file mode 100644 index 0000000000..f70f3c84ba --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/format_conversion/mod.rs @@ -0,0 +1,2 @@ +pub mod fbx_to_gltf; +pub mod process_single_format_conversion_job; \ No newline at end of file diff --git a/crates/service/job/inference_job/src/job/job_types/format_conversion/process_single_format_conversion_job.rs b/crates/service/job/inference_job/src/job/job_types/format_conversion/process_single_format_conversion_job.rs new file mode 100644 index 0000000000..c2e1c9ed12 --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/format_conversion/process_single_format_conversion_job.rs @@ -0,0 +1,60 @@ +use anyhow::anyhow; +use log::error; + +use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; +use mysql_queries::queries::media_files::get_media_file_for_inference::get_media_file_for_inference; +use tokens::tokens::media_files::MediaFileToken; + +use crate::job::job_loop::job_success_result::JobSuccessResult; +use crate::job::job_loop::process_single_job_error::ProcessSingleJobError; +use crate::job::job_types::format_conversion::fbx_to_gltf; +use crate::job::job_types::format_conversion::fbx_to_gltf::process_job::FbxToGltfJobArgs; +use crate::job_dependencies::JobDependencies; + +// NB: There will likely be lots of other format conversions for us to run, and there's probably lots of shared code between them. +// This enum lets us reuse the shared portions in this function before dispatching to the special parts. +#[derive(Debug, Copy, Clone)] +pub enum FormatConversionType { + FbxToGltf, +} + +pub async fn process_single_format_conversion_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob, conversion_type: FormatConversionType) -> Result { + + let maybe_media_file_token = job.maybe_input_source_token + .as_deref() + .map(|token| MediaFileToken::new_from_str(token)); + + let media_file_token = match maybe_media_file_token { + None => return Err(ProcessSingleJobError::Other(anyhow!("no associated media file for format conversion job: {:?}", job.inference_job_token))), + Some(token) => token, + }; + + let maybe_media_file_result = + get_media_file_for_inference(&media_file_token, &job_dependencies.db.mysql_pool).await; + + let media_file = match maybe_media_file_result { + Ok(Some(media_file)) => media_file, + Ok(None) => { + error!("no media file record found for token: {:?}", media_file_token); + return Err(ProcessSingleJobError::Other(anyhow!("no media file record found for token: {:?}", media_file_token))); + } + Err(err) => { + error!("error fetching media file record from db: {:?}", err); + return Err(ProcessSingleJobError::Other(err)); + } + }; + + let job_success_result = match conversion_type { + FormatConversionType::FbxToGltf => { + fbx_to_gltf::process_job::process_job(FbxToGltfJobArgs { + job_dependencies, + job, + media_file: &media_file, + }).await? + } + }; + + Ok(job_success_result) +} + + diff --git a/crates/service/job/inference_job/src/job/job_types/mod.rs b/crates/service/job/inference_job/src/job/job_types/mod.rs index 64b888e0d8..3757b35532 100644 --- a/crates/service/job/inference_job/src/job/job_types/mod.rs +++ b/crates/service/job/inference_job/src/job/job_types/mod.rs @@ -5,3 +5,4 @@ pub(crate) mod videofilter; pub mod image_generation; pub(crate) mod mocap; pub(crate) mod workflow; +pub mod format_conversion;