cleanup job

This commit is contained in:
Brandon Thomas committed 2024-07-15 22:54:22 -04:00
1 parent 59c4574d4d
commit 93ce89e060
7 files changed
+16 -37

No files matched your search

@@ -1,8 +0,0 @@
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use crate::job_dependencies::JobDependencies;
pub struct ComfyProcessJobArgs<'a> {
pub job_dependencies: &'a JobDependencies,
pub job: &'a AvailableInferenceJob,
}
@@ -1,15 +1,15 @@
use errors::anyhow;
use mysql_queries::payloads::generic_inference_args::{generic_inference_args::PolymorphicInferenceArgs, workflow_payload::WorkflowArgs};
use mysql_queries::payloads::generic_inference_args::generic_inference_args::GenericInferenceArgs;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_process_job_args::ComfyProcessJobArgs;
pub fn get_workflow_args_from_job(
args: &ComfyProcessJobArgs<'_>
job: &AvailableInferenceJob,
) -> Result<WorkflowArgs, ProcessSingleJobError> {
let inference_args = args.job.maybe_inference_args
let inference_args = job.maybe_inference_args
.as_ref()
.map(|args: &GenericInferenceArgs| args.args.as_ref())
.flatten();
@@ -1,4 +1,3 @@
pub mod comfy_process_job_args;
pub mod comfy_ui_dependencies;
pub mod get_workflow_args_from_job;
pub mod process_single_wf_job;
@@ -2,7 +2,6 @@ use mysql_queries::queries::generic_inference::job::list_available_generic_infer
use crate::job::job_loop::job_success_result::JobSuccessResult;
use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
use crate::job::job_types::workflow::comfy_process_job_args::ComfyProcessJobArgs;
use crate::job::job_types::workflow::get_workflow_args_from_job::get_workflow_args_from_job;
use crate::job::job_types::workflow::upload_workflow;
use crate::job::job_types::workflow::video_style_transfer;
@@ -13,19 +12,14 @@ pub async fn process_single_wf_job(
job: &AvailableInferenceJob
) -> Result<JobSuccessResult, ProcessSingleJobError> {
let args = ComfyProcessJobArgs {
job_dependencies,
job,
};
let workflow_args = get_workflow_args_from_job(&args)?;
let workflow_args = get_workflow_args_from_job(&job)?;
let job_success_result = match workflow_args.maybe_google_drive_link {
Some(_link) => {
upload_workflow::upload_prompt::upload_prompt(args).await?
upload_workflow::upload_prompt::upload_prompt(job_dependencies, job).await?
}
None => {
video_style_transfer::process_job::process_job(args).await?
video_style_transfer::process_job::process_job(job_dependencies, job).await?
}
};
@@ -11,6 +11,7 @@ use enums::by_table::model_weights::weights_types::WeightsType;
use enums::common::visibility::Visibility;
use filesys::file_exists::file_exists;
use google_drive_common::google_drive_download_command::GoogleDriveDownloadCommand;
use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
use mysql_queries::queries::model_weights::create::create_weight;
use mysql_queries::queries::model_weights::create::create_weight::CreateModelWeightsArgs;
use tokens::tokens::model_weights::ModelWeightToken;
@@ -18,15 +19,13 @@ use tokens::tokens::users::UserToken;
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::workflow::comfy_process_job_args::ComfyProcessJobArgs;
use crate::job::job_types::workflow::get_workflow_args_from_job::get_workflow_args_from_job;
use crate::job_dependencies::JobDependencies;
pub async fn upload_prompt(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResult, ProcessSingleJobError>{
let job = args.job;
let deps = args.job_dependencies;
pub async fn upload_prompt(deps: &JobDependencies, job: &AvailableInferenceJob) -> Result<JobSuccessResult, ProcessSingleJobError>{
let mysql_pool = &deps.db.mysql_pool;
let wf_args = get_workflow_args_from_job(&args)?;
let wf_args = get_workflow_args_from_job(&job)?;
let title = match wf_args.maybe_title {
Some(val) => {
@@ -86,7 +85,7 @@ pub async fn upload_prompt(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessRe
info!("Downloading {}", download_url);
let work_temp_dir = format!("workflow_upload_{}", job.id.0);
let work_temp_dir = args.job_dependencies.fs.scoped_temp_dir_creator_for_work
let work_temp_dir = deps.fs.scoped_temp_dir_creator_for_work
.new_tempdir(&work_temp_dir)
.map_err(|e| ProcessSingleJobError::from_io_error(e))?;
@@ -43,7 +43,6 @@ use videos::ffprobe_get_dimensions::ffprobe_get_dimensions;
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::workflow::comfy_process_job_args::ComfyProcessJobArgs;
use crate::job::job_types::workflow::video_style_transfer::steps::check_and_validate_job::check_and_validate_job;
use crate::job::job_types::workflow::video_style_transfer::steps::download_global_ipa_image::{download_global_ipa_image, DownloadGlobalIpaImageArgs};
use crate::job::job_types::workflow::video_style_transfer::steps::download_input_videos::{download_input_videos, DownloadInputVideoArgs};
@@ -60,18 +59,15 @@ use crate::job_dependencies::JobDependencies;
use crate::util::common_commands::ffmpeg_audio_replace_args::FfmpegAudioReplaceArgs;
use crate::util::common_commands::ffmpeg_logo_watermark_command::WatermarkArgs;
pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResult, ProcessSingleJobError> {
let job = args.job;
let deps = args.job_dependencies;
pub async fn process_job(deps: &JobDependencies, job: &AvailableInferenceJob) -> Result<JobSuccessResult, ProcessSingleJobError> {
let mut job_progress_reporter = args.job_dependencies
let mut job_progress_reporter = deps
.clients
.job_progress_reporter
.new_generic_inference(job.inference_job_token.as_str())
.map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?;
let comfy_deps = args
.job_dependencies
let comfy_deps = deps
.job
.job_specific_dependencies
.maybe_comfy_ui_dependencies
@@ -88,7 +84,7 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
let work_temp_dir = format!("temp_comfy_inference{}", 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
let work_temp_dir = deps
.fs
.scoped_temp_dir_creator_for_work
.new_tempdir(&work_temp_dir)
@@ -492,7 +488,7 @@ pub async fn process_job(args: ComfyProcessJobArgs<'_>) -> Result<JobSuccessResu
info!("Result video media token: {:?}", &media_file_token);
info!("Job {:?} complete success!", args.job.id);
info!("Job {:?} complete success!", job.id);
Ok(JobSuccessResult {
maybe_result_entity: Some(ResultEntity {
@@ -30,7 +30,6 @@ use tokens::tokens::prompts::PromptToken;
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::workflow::comfy_process_job_args::ComfyProcessJobArgs;
use crate::job::job_types::workflow::comfy_ui_dependencies::ComfyDependencies;
use crate::job::job_types::workflow::video_style_transfer::steps::check_and_validate_job::JobArgs;
use crate::job::job_types::workflow::video_style_transfer::util::video_pathing::VideoPathing;