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 3865ef2985..e23b87d34a 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 @@ -40,6 +40,9 @@ pub enum InferenceCategory { #[serde(rename = "format_conversion")] FormatConversion, + + #[serde(rename = "convert_bvh_to_workflow")] + ConvertBvhToWorkflow, } // TODO(bt, 2022-12-21): This desperately needs MySQL integration tests! @@ -58,6 +61,7 @@ impl InferenceCategory { Self::Mocap => "mocap", Self::Workflow => "workflow", Self::FormatConversion => "format_conversion", + Self::ConvertBvhToWorkflow => "convert_bvh_to_workflow", } } @@ -71,6 +75,7 @@ impl InferenceCategory { "mocap" => Ok(Self::Mocap), "workflow" => Ok(Self::Workflow), "format_conversion" => Ok(Self::FormatConversion), + "convert_bvh_to_workflow" => Ok(Self::ConvertBvhToWorkflow), _ => Err(format!("invalid value: {:?}", value)), } } @@ -87,6 +92,7 @@ impl InferenceCategory { Self::Mocap, Self::Workflow, Self::FormatConversion, + Self::ConvertBvhToWorkflow, ]) } } @@ -109,6 +115,7 @@ mod tests { assert_serialization(InferenceCategory::Mocap, "mocap"); assert_serialization(InferenceCategory::Workflow, "workflow"); assert_serialization(InferenceCategory::FormatConversion, "format_conversion"); + assert_serialization(InferenceCategory::ConvertBvhToWorkflow, "convert_bvh_to_workflow"); } #[test] @@ -121,6 +128,7 @@ mod tests { 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::ConvertBvhToWorkflow.to_str(), "convert_bvh_to_workflow"); } #[test] @@ -133,6 +141,7 @@ mod tests { 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_bvh_to_workflow").unwrap(), InferenceCategory::ConvertBvhToWorkflow); } #[test] @@ -148,6 +157,7 @@ mod tests { 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::ConvertBvhToWorkflow)); assert_eq!(variants.pop_first(), None); // Generated check diff --git a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_job_type.rs b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_job_type.rs index 0445c145a8..a654f0e36e 100644 --- a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_job_type.rs +++ b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_job_type.rs @@ -15,6 +15,10 @@ use strum::EnumIter; #[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize, Default)] #[serde(rename_all = "snake_case")] pub enum InferenceJobType { + /// Job that converts bevy to workflow files + #[serde(rename = "bevy_to_workflow")] + BevyToWorkflow, + /// Jobs that run ComfyUI workflows ComfyUi, @@ -64,6 +68,7 @@ impl_mysql_enum_coders!(InferenceJobType); impl InferenceJobType { pub fn to_str(&self) -> &'static str { match self { + Self::BevyToWorkflow => "bevy_to_workflow", Self::ComfyUi => "comfy_ui", Self::ConvertFbxToGltf => "convert_fbx_gltf", Self::MocapNet => "mocap_net", @@ -80,6 +85,7 @@ impl InferenceJobType { pub fn from_str(value: &str) -> Result { match value { + "bevy_to_workflow" => Ok(Self::BevyToWorkflow), "comfy_ui" => Ok(Self::ComfyUi), "convert_fbx_gltf" => Ok(Self::ConvertFbxToGltf), "mocap_net" => Ok(Self::MocapNet), @@ -99,6 +105,7 @@ impl InferenceJobType { // NB: BTreeSet is sorted // NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc. BTreeSet::from([ + Self::BevyToWorkflow, Self::ComfyUi, Self::ConvertFbxToGltf, Self::MocapNet, @@ -129,6 +136,7 @@ mod tests { #[test] fn test_serialization() { + assert_serialization(InferenceJobType::BevyToWorkflow, "bevy_to_workflow"); assert_serialization(InferenceJobType::ComfyUi, "comfy_ui"); assert_serialization(InferenceJobType::ConvertFbxToGltf, "convert_fbx_gltf"); assert_serialization(InferenceJobType::MocapNet, "mocap_net"); @@ -144,6 +152,7 @@ mod tests { #[test] fn to_str() { + assert_eq!(InferenceJobType::BevyToWorkflow.to_str(), "bevy_to_workflow"); assert_eq!(InferenceJobType::ComfyUi.to_str(), "comfy_ui"); assert_eq!(InferenceJobType::ConvertFbxToGltf.to_str(), "convert_fbx_gltf"); assert_eq!(InferenceJobType::MocapNet.to_str(), "mocap_net"); @@ -159,6 +168,7 @@ mod tests { #[test] fn from_str() { + assert_eq!(InferenceJobType::from_str("bevy_to_workflow").unwrap(), InferenceJobType::BevyToWorkflow); assert_eq!(InferenceJobType::from_str("comfy_ui").unwrap(), InferenceJobType::ComfyUi); assert_eq!(InferenceJobType::from_str("convert_fbx_gltf").unwrap(), InferenceJobType::ConvertFbxToGltf); assert_eq!(InferenceJobType::from_str("mocap_net").unwrap(), InferenceJobType::MocapNet); @@ -176,7 +186,8 @@ mod tests { fn all_variants() { // Static check let mut variants = InferenceJobType::all_variants(); - assert_eq!(variants.len(), 11); + assert_eq!(variants.len(), 12); + assert_eq!(variants.pop_first(), Some(InferenceJobType::BevyToWorkflow)); assert_eq!(variants.pop_first(), Some(InferenceJobType::ComfyUi)); assert_eq!(variants.pop_first(), Some(InferenceJobType::ConvertFbxToGltf)); assert_eq!(variants.pop_first(), Some(InferenceJobType::MocapNet)); diff --git a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_model_type.rs b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_model_type.rs index cc0c2f754d..4f68153e05 100644 --- a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_model_type.rs +++ b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_model_type.rs @@ -44,6 +44,8 @@ pub enum InferenceModelType { /// A job that turns "FBX" game engine files into "GLTF" files (Bevy-compatible). #[serde(rename = "convert_fbx_gltf")] ConvertFbxToGltf, + #[serde(rename = "bvh_to_workflow")] + BvhToWorkflow } // TODO(bt, 2022-12-21): This desperately needs MySQL integration tests! @@ -66,6 +68,7 @@ impl InferenceModelType { Self::StyleTTS2 => "styletts2", Self::ComfyUi => "comfy_ui", Self::ConvertFbxToGltf => "convert_fbx_gltf", + Self::BvhToWorkflow => "bvh_to_workflow", } } @@ -83,6 +86,7 @@ impl InferenceModelType { "styletts2" => Ok(Self::StyleTTS2), "comfy_ui" => Ok(Self::ComfyUi), "convert_fbx_gltf" => Ok(Self::ConvertFbxToGltf), + "bvh_to_workflow" => Ok(Self::BvhToWorkflow), _ => Err(format!("invalid value: {:?}", value)), } } @@ -103,6 +107,7 @@ impl InferenceModelType { Self::StyleTTS2, Self::ComfyUi, Self::ConvertFbxToGltf, + Self::BvhToWorkflow, ]) } } @@ -129,6 +134,7 @@ mod tests { assert_serialization(InferenceModelType::ComfyUi, "comfy_ui"); assert_serialization(InferenceModelType::StyleTTS2, "styletts2"); assert_serialization(InferenceModelType::ConvertFbxToGltf, "convert_fbx_gltf"); + assert_serialization(InferenceModelType::BvhToWorkflow, "bvh_to_workflow"); } #[test] @@ -145,6 +151,7 @@ mod tests { assert_eq!(InferenceModelType::StyleTTS2.to_str(), "styletts2"); assert_eq!(InferenceModelType::ComfyUi.to_str(), "comfy_ui"); assert_eq!(InferenceModelType::ConvertFbxToGltf.to_str(), "convert_fbx_gltf"); + assert_eq!(InferenceModelType::BvhToWorkflow.to_str(), "bvh_to_workflow"); } #[test] @@ -161,6 +168,7 @@ mod tests { assert_eq!(InferenceModelType::from_str("styletts2").unwrap(), InferenceModelType::StyleTTS2); assert_eq!(InferenceModelType::from_str("comfy_ui").unwrap(), InferenceModelType::ComfyUi); assert_eq!(InferenceModelType::from_str("convert_fbx_gltf").unwrap(), InferenceModelType::ConvertFbxToGltf); + assert_eq!(InferenceModelType::from_str("bvh_to_workflow").unwrap(), InferenceModelType::BvhToWorkflow); } #[test] @@ -180,6 +188,7 @@ mod tests { assert_eq!(variants.pop_first(), Some(InferenceModelType::ComfyUi)); assert_eq!(variants.pop_first(), Some(InferenceModelType::StyleTTS2)); assert_eq!(variants.pop_first(), Some(InferenceModelType::ConvertFbxToGltf)); + assert_eq!(variants.pop_first(), Some(InferenceModelType::BvhToWorkflow)); assert_eq!(variants.pop_first(), None); // Generated check 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 8e6592789e..3037bf3f09 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 @@ -58,6 +58,10 @@ pub enum InferenceCategoryAbbreviated { #[serde(rename = "fc")] // NB: DO NOT CHANGE. It could break live jobs. Renamed to be fewer bytes. #[serde(alias = "format_conversion")] FormatConversion, + + #[serde(rename = "bw")] // NB: DO NOT CHANGE. It could break live jobs. Renamed to be fewer bytes. + #[serde(alias = "convert_bvh_to_workflow")] + ConvertBvhToWorkflow, } #[derive(Serialize, Deserialize, Clone, Copy, Debug)] @@ -143,6 +147,7 @@ impl InferenceCategoryAbbreviated { InferenceCategory::Mocap => Self::Mocap, InferenceCategory::Workflow => Self::Workflow, InferenceCategory::FormatConversion => Self::FormatConversion, + InferenceCategory::ConvertBvhToWorkflow => Self::ConvertBvhToWorkflow, } } @@ -156,6 +161,7 @@ impl InferenceCategoryAbbreviated { Self::Mocap => InferenceCategory::Mocap, Self::Workflow => InferenceCategory::Workflow, Self::FormatConversion => InferenceCategory::FormatConversion, + Self::ConvertBvhToWorkflow => InferenceCategory::ConvertBvhToWorkflow, } } } 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 94829199cd..73dfae66bf 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 @@ -183,6 +183,7 @@ fn raw_record_to_public_result(record: RawGenericInferenceJobStatus) -> GenericI InferenceCategory::Mocap => Some("Mocap"), InferenceCategory::Workflow => Some("Workflow"), InferenceCategory::FormatConversion => Some("format conversion"), + InferenceCategory::ConvertBvhToWorkflow => Some("BVH to Workflow"), }; // NB: A bit of a hack. We store TTS results with a full path. @@ -196,6 +197,7 @@ fn raw_record_to_public_result(record: RawGenericInferenceJobStatus) -> GenericI 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::FormatConversion => (true, record.maybe_media_file_public_bucket_directory_hash.as_deref()), + InferenceCategory::ConvertBvhToWorkflow => (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/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 82a5bb5e1e..029b4b564b 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,6 +13,7 @@ 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::bevy_to_workflow::process_single_bevy_to_workflow_conversion_job::process_single_bevy_to_workflow_conversion_job; use crate::job::job_types::format_conversion::process_single_format_conversion_job::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; @@ -202,6 +203,9 @@ async fn do_process_single_job( InferenceCategory::FormatConversion => { process_single_format_conversion_job(job_dependencies, job).await? } + InferenceCategory::ConvertBvhToWorkflow => { + process_single_bevy_to_workflow_conversion_job(job_dependencies, job).await? + } }; let maybe_entity_type = job_success_result.maybe_result_entity diff --git a/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/command_args.rs b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/command_args.rs new file mode 100644 index 0000000000..3705894932 --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/command_args.rs @@ -0,0 +1,23 @@ +use std::path::Path; + +use filesys::path_to_string::path_to_string; +use subprocess_common::command_runner::command_args::CommandArgs; + +pub struct BvhToWorkflowCommandArgs<'a> { + pub input_file: &'a Path, + pub output_directory: &'a Path, +} + +impl CommandArgs for BvhToWorkflowCommandArgs<'_> { + fn to_command_string(&self) -> String { + let mut command = String::new(); + + command.push_str(" -i "); + command.push_str(&path_to_string(self.input_file)); + + command.push_str(" -o "); + command.push_str(&path_to_string(self.output_directory)); + + command + } +} diff --git a/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/dependencies.rs b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/dependencies.rs new file mode 100644 index 0000000000..3aaa5c9bd3 --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/dependencies.rs @@ -0,0 +1,46 @@ +use std::path::PathBuf; + +use errors::AnyhowResult; +use subprocess_common::command_runner::command_runner::CommandRunner; +use subprocess_common::command_runner::env_var_policy::EnvVarPolicy; +use subprocess_common::docker_options::{DockerFilesystemMount, DockerOptions}; +use subprocess_common::executable_or_command::ExecutableOrShellCommand; + +pub struct BvhToWorkflowDependencies { + pub command_runner: CommandRunner, +} + +impl BvhToWorkflowDependencies { + pub fn setup() -> AnyhowResult { + let maybe_execution_directory = easyenv::get_env_pathbuf_optional("BVH2WORKFLOW_DIRECTORY"); + + let executable_or_command = easyenv::get_env_string_required("BVH2WORKFLOW_EXECUTABLE_OR_COMMAND")?; + + let executable_or_command = if executable_or_command.contains(" ") { + ExecutableOrShellCommand::BashShellCommand(executable_or_command) + } else { + ExecutableOrShellCommand::Executable(PathBuf::from(executable_or_command)) + }; + + let maybe_execution_timeout = easyenv::get_env_duration_seconds_optional("BVH2WORKFLOW_TIMEOUT_SECONDS"); + + let maybe_docker_options = easyenv::get_env_string_optional("BVH2WORKFLOW_DOCKER_IMAGE") + .map(|image_name| DockerOptions { + image_name, + maybe_bind_mount: Some(DockerFilesystemMount::tmp_to_tmp()), + maybe_environment_variables: None, + maybe_gpu: None, + }); + + Ok(Self { + command_runner: CommandRunner { + executable_or_command, + maybe_execution_directory, + env_var_policy: EnvVarPolicy::CopyNone, + maybe_virtual_env_activation_command: None, + maybe_docker_options, + maybe_execution_timeout, + } + }) + } +} diff --git a/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/mod.rs b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/mod.rs new file mode 100644 index 0000000000..90e87e5770 --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/mod.rs @@ -0,0 +1,3 @@ +pub mod process_job; +pub mod command_args; +pub mod dependencies; \ No newline at end of file diff --git a/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/process_job.rs b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/process_job.rs new file mode 100644 index 0000000000..7d2126707e --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/bvh_to_workflow/process_job.rs @@ -0,0 +1,290 @@ +use std::fs::read_to_string; +use std::time::Instant; + +use anyhow::anyhow; +use log::{error, info, warn}; +use buckets::public::media_files::bucket_file_path::MediaFileBucketPath; + +use buckets::public::media_uploads::bucket_file_path::MediaUploadOriginalFilePath; +use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType; +use enums::by_table::generic_synthetic_ids::id_category::IdCategory; +use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory; +use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory; +use enums::by_table::media_files::media_file_type::MediaFileType; +use filesys::check_file_exists::check_file_exists; +use filesys::file_size::file_size; +use filesys::safe_delete_temp_directory::safe_delete_temp_directory; +use filesys::safe_delete_temp_file::safe_delete_temp_file; +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::create::insert_media_file_generic::{insert_media_file_generic, InsertArgs}; +use mysql_queries::queries::media_files::get_media_file_for_inference::MediaFileForInference; +use subprocess_common::command_runner::command_runner_args::{FileOrCreate, RunAsSubprocessArgs}; + +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::bevy_to_workflow::bvh_to_workflow::command_args::BvhToWorkflowCommandArgs; +use crate::job_dependencies::JobDependencies; +use crate::util::maybe_download_file_from_bucket::{maybe_download_file_from_bucket, MaybeDownloadArgs}; + +const BUCKET_FILE_EXTENSION : &str = ".mp4"; +const BUCKET_FRAMES_FILE_EXTENSION : &str = ".zip"; + +pub struct BvhToWorkflowJobArgs<'a> { + pub job_dependencies: &'a JobDependencies, + pub job: &'a AvailableInferenceJob, + pub media_file: &'a MediaFileForInference, +} + +pub async fn process_job(args: BvhToWorkflowJobArgs<'_>) -> 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)))?; + + let model_dependencies = args + .job_dependencies + .job + .job_specific_dependencies + .maybe_convert_bvh_to_workflow_dependencies + .as_ref() + .ok_or_else(|| ProcessSingleJobError::JobSystemMisconfiguration(Some("missing BvhToWorkflow dependencies".to_string())))?; + + // ==================== 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_file_fs_path = work_temp_dir.path().join("original.bvh"); + + let media_file_bucket_path = MediaFileBucketPath::from_object_hash( + &media_file.public_bucket_directory_hash, + media_file.maybe_public_bucket_prefix.as_deref(), + media_file.maybe_public_bucket_extension.as_deref()); + + let bucket_object_path = media_file_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_file_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_file_fs_path + }; + + // ==================== SETUP FOR CONVERSION ==================== // + + job_progress_reporter.log_status("running conversion") + .map_err(|e| ProcessSingleJobError::Other(e))?; + + + // ==================== RUN INFERENCE SCRIPT ==================== // + + let stderr_output_file = work_temp_dir.path().join("stderr.txt"); + let output_directory = work_temp_dir.path().join("output"); + let output_directory_actual = work_temp_dir.path(); + + let execution_start_time = Instant::now(); + + let command_exit_status = { + model_dependencies + .command_runner + .run_with_subprocess(RunAsSubprocessArgs { + args: Box::new(&BvhToWorkflowCommandArgs { + input_file: &original_media_upload_fs_path, + output_directory: &output_directory, + }), + maybe_stderr_output_file: Some(FileOrCreate::NewFileWithName(&stderr_output_file)), + maybe_stdout_output_file: None, + }) + }; + + let execution_duration = Instant::now().duration_since(execution_start_time); + + info!("Execution took duration to complete: {:?}", &execution_duration); + + if !command_exit_status.is_success() { + error!("Execution 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); + } + + safe_delete_temp_file(&original_media_upload_fs_path); + safe_delete_temp_directory(&output_directory_actual); + safe_delete_temp_directory(&work_temp_dir); + + return Err(error); + } + + // ==================== CHECK ALL FILES EXIST AND GET METADATA ==================== // + + let output_video_file = output_directory_actual.join("output.mp4"); + let output_frames_zip_file = output_directory_actual.join("output.zip"); + // NB: The actual name of the file will be this: + + info!("Checking that output files exist..."); + + check_file_exists(&output_video_file).map_err(|e| ProcessSingleJobError::Other(e))?; + check_file_exists(&output_frames_zip_file).map_err(|e| ProcessSingleJobError::Other(e))?; + + info!("Interrogating result file properties..."); + + let file_size_bytes = file_size(&output_video_file) + .map_err(|err| ProcessSingleJobError::Other(err))?; + + let maybe_mimetype = get_mimetype_for_file(&output_video_file) + .map_err(|err| ProcessSingleJobError::from_io_error(err))? + .map(|mime| mime.to_string()); + + info!("Calculating sha256..."); + + let file_checksum = sha256_hash_file(&output_video_file) + .map_err(|err| { + ProcessSingleJobError::Other(anyhow!("Error hashing file: {:?}", err)) + })?; + + // ==================== UPLOAD AUDIO TO BUCKET ==================== // + + job_progress_reporter.log_status("uploading result") + .map_err(|e| ProcessSingleJobError::Other(e))?; + + let result_bucket_location = MediaFileBucketPath::generate_new( + None, + Some(BUCKET_FILE_EXTENSION)); + + let result_bucket_object_pathbuf = result_bucket_location.to_full_object_pathbuf(); + + info!("Destination bucket path: {:?}", &result_bucket_object_pathbuf); + + info!("Uploading media file..."); + + args.job_dependencies.buckets.public_bucket_client.upload_filename_with_content_type( + &result_bucket_object_pathbuf, + &output_video_file, + "video/mp4") + .await + .map_err(|e| ProcessSingleJobError::Other(e))?; + + let result_bucket_location = MediaFileBucketPath::from_object_hash(&result_bucket_location.get_object_hash(), + None, + Some(".zip")); + + let result_bucket_object_pathbuf = result_bucket_location.to_full_object_pathbuf(); + + info!("Destination bucket path: {:?}", &result_bucket_object_pathbuf); + + info!("Uploading media file..."); + + args.job_dependencies.buckets.public_bucket_client.upload_filename_with_content_type( + &result_bucket_object_pathbuf, + &output_frames_zip_file, + "application/zip") + .await + .map_err(|e| ProcessSingleJobError::Other(e))?; + + safe_delete_temp_file(&output_video_file); + safe_delete_temp_file(&output_frames_zip_file); + safe_delete_temp_directory(&output_directory_actual); + + // ==================== 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 record..."); + + let (inference_result_token, id) = insert_media_file_generic(InsertArgs { + pool: &args.job_dependencies.db.mysql_pool, + job: &job, + media_type: MediaFileType::Gltf, + origin_category: MediaFileOriginCategory::Processed, + origin_product_category: MediaFileOriginProductCategory::Mocap, + maybe_origin_model_type: None, + maybe_origin_model_token: None, + maybe_origin_filename: None, + maybe_mime_type: maybe_mimetype.as_deref(), + file_size_bytes, + maybe_duration_millis: None, + maybe_audio_encoding: None, + maybe_video_encoding: None, + maybe_frame_width: None, + maybe_frame_height: None, + public_bucket_directory_hash: result_bucket_location.get_object_hash(), + maybe_public_bucket_prefix: result_bucket_location.get_optional_prefix(), + maybe_public_bucket_extension: result_bucket_location.get_optional_extension(), + extra_file_modification_info: None, + maybe_creator_file_synthetic_id_category: IdCategory::MediaFile, + maybe_creator_category_synthetic_id_category: IdCategory::MocapResult, + maybe_mod_user_token: None, + is_batch_generated: false, + is_generated_on_prem: false, + checksum_sha2: &file_checksum, + generated_by_worker: Some(&args.job_dependencies.job.info.container.hostname), + generated_by_cluster: Some(&args.job_dependencies.job.info.container.cluster_name), + }) + .await + .map_err(|e| ProcessSingleJobError::Other(e))?; + + info!("VC Done."); + + //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, executed successfully, and uploaded. Saved 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(), + }), + inference_duration: execution_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/bevy_to_workflow/mod.rs b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/mod.rs new file mode 100644 index 0000000000..a03c4d77c9 --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/mod.rs @@ -0,0 +1,2 @@ +pub mod bvh_to_workflow; +pub mod process_single_bevy_to_workflow_conversion_job; \ No newline at end of file diff --git a/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/process_single_bevy_to_workflow_conversion_job.rs b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/process_single_bevy_to_workflow_conversion_job.rs new file mode 100644 index 0000000000..fa34aded0f --- /dev/null +++ b/crates/service/job/inference_job/src/job/job_types/bevy_to_workflow/process_single_bevy_to_workflow_conversion_job.rs @@ -0,0 +1,50 @@ +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::bevy_to_workflow::bvh_to_workflow; +use crate::job::job_types::bevy_to_workflow::bvh_to_workflow::process_job::BvhToWorkflowJobArgs; +use crate::job_dependencies::JobDependencies; + + +pub async fn process_single_bevy_to_workflow_conversion_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> 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 = bvh_to_workflow::process_job::process_job(BvhToWorkflowJobArgs { + 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 3757b35532..b02a9ccf9f 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 @@ -6,3 +6,4 @@ pub mod image_generation; pub(crate) mod mocap; pub(crate) mod workflow; pub mod format_conversion; +pub mod bevy_to_workflow; diff --git a/crates/service/job/inference_job/src/job_specific_dependencies.rs b/crates/service/job/inference_job/src/job_specific_dependencies.rs index 7287cebf0b..758110d284 100644 --- a/crates/service/job/inference_job/src/job_specific_dependencies.rs +++ b/crates/service/job/inference_job/src/job_specific_dependencies.rs @@ -1,5 +1,6 @@ use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType; use errors::AnyhowResult; +use crate::job::job_types::bevy_to_workflow::bvh_to_workflow::dependencies::BvhToWorkflowDependencies; use crate::job::job_types::format_conversion::fbx_to_gltf::dependencies::FbxToGltfDependencies; use crate::job::job_types::image_generation::sd::stable_diffusion_dependencies::StableDiffusionDependencies; @@ -28,6 +29,7 @@ pub struct JobSpecificDependencies { pub maybe_styletts2_dependencies: Option, pub maybe_comfy_ui_dependencies: Option, pub maybe_convert_fbx_to_gltf_dependencies: Option, + pub maybe_convert_bvh_to_workflow_dependencies: Option, } impl JobSpecificDependencies { @@ -45,6 +47,7 @@ impl JobSpecificDependencies { let mut maybe_styletts2_dependencies = None; let mut maybe_comfy_ui_dependencies = None; let mut maybe_convert_fbx_to_gltf_dependencies = None; + let mut maybe_convert_bvh_to_workflow_dependencies = None; if scoped_execution.can_run_job(InferenceModelType::RvcV2) { print_with_space("Setting RVCv2 dependencies..."); @@ -106,6 +109,11 @@ impl JobSpecificDependencies { maybe_convert_fbx_to_gltf_dependencies = Some(FbxToGltfDependencies::setup()?); } + if scoped_execution.can_run_job(InferenceModelType::BvhToWorkflow) { + print_with_space("Setting ConvertBvhToWorkflow dependencies..."); + maybe_convert_bvh_to_workflow_dependencies = Some(BvhToWorkflowDependencies::setup()?); + } + Ok(JobSpecificDependencies { maybe_rvc_v2_dependencies, maybe_sad_talker_dependencies, @@ -119,6 +127,7 @@ impl JobSpecificDependencies { maybe_styletts2_dependencies, maybe_comfy_ui_dependencies, maybe_convert_fbx_to_gltf_dependencies, + maybe_convert_bvh_to_workflow_dependencies, }) } } diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/conversion/enqueue_bvh_to_workflow_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/conversion/enqueue_bvh_to_workflow_handler.rs new file mode 100644 index 0000000000..af40f101a0 --- /dev/null +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/conversion/enqueue_bvh_to_workflow_handler.rs @@ -0,0 +1,218 @@ +#![forbid(unused_mut)] + +use std::fmt::Debug; +use std::sync::Arc; + +use actix_web::{HttpRequest, HttpResponse, web}; +use actix_web::error::ResponseError; +use actix_web::http::StatusCode; +use log::warn; +use serde::Deserialize; +use serde::Serialize; +use utoipa::ToSchema; + +use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory; +use enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType; +use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType; +use enums::common::visibility::Visibility; +use http_server_common::request::get_request_header_optional::get_request_header_optional; +use http_server_common::request::get_request_ip::get_request_ip; +use mysql_queries::payloads::generic_inference_args::generic_inference_args::{ + GenericInferenceArgs, + InferenceCategoryAbbreviated, +}; +use mysql_queries::queries::generic_inference::web::insert_generic_inference_job::{ + insert_generic_inference_job, + InsertGenericInferenceArgs, +}; +use tokens::tokens::generic_inference_jobs::InferenceJobToken; +use tokens::tokens::media_files::MediaFileToken; +use tokens::tokens::users::UserToken; + +use crate::configs::plans::get_correct_plan_for_session::get_correct_plan_for_session; +use crate::http_server::web_utils::response_error_helpers::to_simple_json_error; +use crate::server_state::ServerState; + +/// Debug requests can get routed to special "debug-only" workers, which can +/// be used to trial new code, run debugging, etc. +const DEBUG_HEADER_NAME: &str = "enable-debug-mode"; + +/// The routing tag header can send workloads to particular k8s hosts. +/// This is useful for catching the live logs or intercepting the job. +const ROUTING_TAG_HEADER_NAME: &str = "routing-tag"; + +#[derive(Deserialize, ToSchema)] +pub struct EnqueueBvhToWorkflowRequest { + // Entropy for idempotency + uuid_idempotency_token: String, + // The existing FBX media file token + media_file_token: MediaFileToken, +} + +#[derive(Serialize, ToSchema)] +pub struct EnqueueBvhToWorkflowRequestSuccessResponse { + pub success: bool, + pub inference_job_token: InferenceJobToken, +} + +#[derive(Debug, ToSchema)] +pub enum EnqueueBvhToWorkflowRequestError { + BadInput(String), + NotAuthorized, + ServerError, + RateLimited, +} + +impl ResponseError for EnqueueBvhToWorkflowRequestError { + fn status_code(&self) -> StatusCode { + match *self { + EnqueueBvhToWorkflowRequestError::BadInput(_) => StatusCode::BAD_REQUEST, + EnqueueBvhToWorkflowRequestError::NotAuthorized => StatusCode::UNAUTHORIZED, + EnqueueBvhToWorkflowRequestError::ServerError => StatusCode::INTERNAL_SERVER_ERROR, + EnqueueBvhToWorkflowRequestError::RateLimited => StatusCode::TOO_MANY_REQUESTS, + } + } + + fn error_response(&self) -> HttpResponse { + let error_reason = match self { + EnqueueBvhToWorkflowRequestError::BadInput(reason) => reason.to_string(), + EnqueueBvhToWorkflowRequestError::NotAuthorized => "unauthorized".to_string(), + EnqueueBvhToWorkflowRequestError::ServerError => "server error".to_string(), + EnqueueBvhToWorkflowRequestError::RateLimited => "rate limited".to_string(), + }; + + to_simple_json_error(&error_reason, self.status_code()) + } +} + +impl std::fmt::Display for EnqueueBvhToWorkflowRequestError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{:?}", self) + } +} + +#[utoipa::path( + post, + path = "/v1/conversion/enqueue_fbx_to_gltf", + responses( + ( + status = 200, + description = "Enqueue BVH To Workflow", + body = EnqueueBvhToWorkflowRequestSuccessResponse, + ), + (status = 400, description = "Bad input", body = EnqueueBvhToWorkflowRequestError), + (status = 401, description = "Not authorized", body = EnqueueBvhToWorkflowRequestError), + (status = 429, description = "Rate limited", body = EnqueueBvhToWorkflowRequestError), + (status = 500, description = "Server error", body = EnqueueBvhToWorkflowRequestError) + ), + params(("request" = EnqueueBvhToWorkflowRequest, description = "Payload for BVH to Workflow Request")) +)] +pub async fn enqueue_bvh_to_workflow_handler( + http_request: HttpRequest, + request: web::Json, + server_state: web::Data> +) -> Result { + let mut maybe_user_token: Option = None; + + let mut mysql_connection = server_state.mysql_pool.acquire().await.map_err(|err| { + warn!("MySql pool error: {:?}", err); + EnqueueBvhToWorkflowRequestError::ServerError + })?; + + // ==================== USER SESSION ==================== // + + let maybe_user_session = server_state.session_checker + .maybe_get_user_session_extended_from_connection(&http_request, &mut mysql_connection).await + .map_err(|e| { + warn!("Session checker error: {:?}", e); + EnqueueBvhToWorkflowRequestError::ServerError + })?; + + if let Some(user_session) = maybe_user_session.as_ref() { + maybe_user_token = Some(UserToken::new_from_str(&user_session.user_token)); + } + + // Plan should handle "first anonymous use" and "investor" cases. + let plan = get_correct_plan_for_session( + server_state.server_environment, + maybe_user_session.as_ref() + ); + + // Separate priority for animation. + let priority_level = plan.web_vc_base_priority_level(); + + // ==================== DEBUG MODE + ROUTING TAG ==================== // + + let is_debug_request = get_request_header_optional(&http_request, DEBUG_HEADER_NAME).is_some(); + + let maybe_routing_tag = get_request_header_optional(&http_request, ROUTING_TAG_HEADER_NAME).map( + |routing_tag| routing_tag.trim().to_string() + ); + + // ==================== BANNED USERS ==================== // + + if let Some(ref user) = maybe_user_session { + if user.role.is_banned { + return Err(EnqueueBvhToWorkflowRequestError::NotAuthorized); + } + } + + // ==================== RATE LIMIT ==================== // + + let rate_limiter = match maybe_user_session { + None => &server_state.redis_rate_limiters.logged_out, + Some(ref _user) => &server_state.redis_rate_limiters.logged_in, + }; + + if let Err(_err) = rate_limiter.rate_limit_request(&http_request) { + return Err(EnqueueBvhToWorkflowRequestError::RateLimited); + } + + let ip_address = get_request_ip(&http_request); + + let maybe_avt_token = server_state.avt_cookie_manager + .get_avt_token_from_request(&http_request); + + let query_result = insert_generic_inference_job(InsertGenericInferenceArgs { + uuid_idempotency_token: &request.uuid_idempotency_token, + inference_category: InferenceCategory::ConvertBvhToWorkflow, + maybe_model_type: Some(InferenceModelType::BvhToWorkflow), + maybe_model_token: None, + job_type: InferenceJobType::BevyToWorkflow, + maybe_input_source_token: Some(&request.media_file_token.as_str()), + maybe_input_source_token_type: None, + maybe_raw_inference_text: None, + maybe_max_duration_seconds: None, + maybe_inference_args: Some(GenericInferenceArgs { + inference_category: Some(InferenceCategoryAbbreviated::ConvertBvhToWorkflow), + args: None + }), + maybe_creator_user_token: maybe_user_token.as_ref(), + maybe_avt_token: maybe_avt_token.as_ref(), + creator_ip_address: &ip_address, + creator_set_visibility: Visibility::Public, + priority_level, + requires_keepalive: true, + is_debug_request, + maybe_routing_tag: maybe_routing_tag.as_deref(), + 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(EnqueueBvhToWorkflowRequestError::ServerError); + } + }; + + let response: EnqueueBvhToWorkflowRequestSuccessResponse = EnqueueBvhToWorkflowRequestSuccessResponse { + success: true, + inference_job_token: job_token, + }; + + let body = serde_json::to_string(&response) + .map_err(|_e| EnqueueBvhToWorkflowRequestError::ServerError)?; + + Ok(HttpResponse::Ok().content_type("application/json").body(body)) +} diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/conversion/mod.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/conversion/mod.rs index 9340f80623..0d33ba90e5 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/conversion/mod.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/conversion/mod.rs @@ -1 +1,2 @@ -pub mod enqueue_fbx_to_gltf_handler; \ No newline at end of file +pub mod enqueue_fbx_to_gltf_handler; +pub mod enqueue_bvh_to_workflow_handler; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status.rs index 613b857530..b5cf73bfd3 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/inference_job/get_inference_job_status.rs @@ -324,6 +324,9 @@ fn record_to_payload( InferenceCategory::FormatConversion => { "".to_string() } + InferenceCategory::ConvertBvhToWorkflow => { + "".to_string() + } }; ResultDetailsResponse { diff --git a/crates/service/web/storyteller_web/src/routes.rs b/crates/service/web/storyteller_web/src/routes.rs index 81f4292803..09199c2dae 100644 --- a/crates/service/web/storyteller_web/src/routes.rs +++ b/crates/service/web/storyteller_web/src/routes.rs @@ -26,6 +26,7 @@ use crate::http_server::endpoints::categories::tts::list_tts_model_assigned_cate use crate::http_server::endpoints::comments::create_comment_handler::create_comment_handler; use crate::http_server::endpoints::comments::delete_comment_handler::delete_comment_handler; use crate::http_server::endpoints::comments::list_comments_handler::list_comments_handler; +use crate::http_server::endpoints::conversion::enqueue_bvh_to_workflow_handler::enqueue_bvh_to_workflow_handler; use crate::http_server::endpoints::conversion::enqueue_fbx_to_gltf_handler::enqueue_fbx_to_gltf_handler; use crate::http_server::endpoints::download_job::enqueue_generic_download::enqueue_generic_download_handler; use crate::http_server::endpoints::download_job::get_generic_upload_job_status::get_generic_download_job_status_handler; @@ -257,6 +258,11 @@ pub fn add_routes (app: App, server_environment: ServerEnvironment) -> .add_post("/v1/conversion/enqueue_fbx_to_gltf", enqueue_fbx_to_gltf_handler) .into_app(); + // =================== BVH from Workflow ==================== + let mut app = RouteBuilder::from_app(app) + .add_post("/v1/conversion/enqueue_bvh_to_workflow", enqueue_bvh_to_workflow_handler) + .into_app(); + // ==================== Stats ==================== let mut app = RouteBuilder::from_app(app)