Merge pull request #293 from storytold/jan-31-mithoo

scaffold for bvh -> workflow job

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

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<Self, String> {
     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));

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

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,
     }
   }
 }

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

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

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
+    }
+}

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<Self> {
+        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,
+            }
+        })
+    }
+}

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;

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<JobSuccessResult, ProcessSingleJobError> {
+  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<u64>,
+  pub mimetype: Option<String>,
+  pub file_size_bytes: u64,
+}

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;

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<JobSuccessResult, ProcessSingleJobError> {
+
+  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)
+}
+
+

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;

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<StyleTTS2Dependencies>,
   pub maybe_comfy_ui_dependencies: Option<ComfyDependencies>,
   pub maybe_convert_fbx_to_gltf_dependencies: Option<FbxToGltfDependencies>,
+  pub maybe_convert_bvh_to_workflow_dependencies: Option<BvhToWorkflowDependencies>,
 }
 
 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,
     })
   }
 }

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<EnqueueBvhToWorkflowRequest>,
+    server_state: web::Data<Arc<ServerState>>
+) -> Result<HttpResponse, EnqueueBvhToWorkflowRequestError> {
+    let mut maybe_user_token: Option<UserToken> = 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))
+}

crates/service/web/storyteller_web/src/http_server/endpoints/conversion/mod.rs

@@ -1 +1,2 @@
-pub mod enqueue_fbx_to_gltf_handler;
+pub mod enqueue_fbx_to_gltf_handler;
+pub mod enqueue_bvh_to_workflow_handler;

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 {

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<T, B> (app: App<T>, 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)