Ok this builds inference job. Fixes still needed

crates/lib/api_clients/cloud_storage/src/remote_file_manager/media_descriptor.rs

@@ -0,0 +1,21 @@
+use crate::remote_file_manager::file_descriptor::FileDescriptor;
+// TODO ASK BRANDON WHAT path should this be?
+const MEDIA_FILE_DIRECTORY: &str = "/media";
+
+pub struct MediaImagePngDescriptor {}
+impl FileDescriptor for MediaImagePngDescriptor {
+    fn remote_directory_path(&self) -> &str {
+        MEDIA_FILE_DIRECTORY
+    }
+    fn get_suffix(&self)->String {
+        "png".to_string()
+    }
+
+    fn get_prefix(&self)->String {
+        "image".to_string()
+    }
+
+    fn is_public(&self) -> bool {
+        true
+    }
+}

crates/lib/api_clients/cloud_storage/src/remote_file_manager/weights_descriptor.rs

@@ -3,7 +3,6 @@ use super::file_descriptor::FileDescriptor;
 pub struct WeightsLoRADescriptor;
 
 const REMOTE_FILE_DIRECTORY: &str = "/weights";
-
 impl FileDescriptor for WeightsLoRADescriptor {
     fn remote_directory_path(&self) -> &str {
         REMOTE_FILE_DIRECTORY
@@ -26,6 +25,7 @@ impl FileDescriptor for WeightsLoRADescriptor {
     }
 }
 
+
 pub struct WeightsSD15Descriptor;
 
 impl FileDescriptor for WeightsSD15Descriptor {

crates/lib/api_clients/google_drive_common/src/google_drive_download_command.rs

@@ -57,6 +57,60 @@ impl GoogleDriveDownloadCommand {
     }
   }
 
+  pub async fn download_file_with_file_name(&self,
+                             download_url: &str,
+                             temp_dir: &TempDir,
+                             file_name:&str) -> AnyhowResult<String>
+  {
+    let temp_dir_path = temp_dir.path()
+        .to_str()
+        .unwrap_or("/tmp")
+        .to_string();
+
+    let temp_filename = random_crockford_token(10);
+    let temp_filename = format!("{}/{}", temp_dir_path, file_name);
+
+    info!("Downloading {} to: {}", download_url, temp_filename);
+
+    let mut command = format!("{} --url \"{}\" --output_filename {}",
+                              &self.download_script,
+                              download_url,
+                              &temp_filename);
+
+    if let Some(venv_activation_script) = self.maybe_venv_activation_script.as_deref() {
+      // NB: "." is source for non-bash shells
+      command = format!(". {} && {}",
+                        venv_activation_script,
+                        &command);
+    }
+
+    if let Some(change_to_directory) = self.maybe_change_to_directory.as_deref() {
+      command = format!("cd {} && {}",
+                        change_to_directory,
+                        &command);
+    }
+
+    if let Some(docker_options) = self.maybe_docker_options.as_ref() {
+      command = docker_options.to_command_string(&command);
+    }
+
+    info!("Running command: {}", command);
+
+    let result = Command::new("sh")
+        .arg("-c")
+        .arg(command)
+        .output()?;
+
+    info!("Downloader Result: {:?}", result);
+
+    if !result.status.success() {
+      let reason = String::from_utf8(result.stderr).unwrap_or("UNKNOWN".to_string());
+      return Err(anyhow!("Failure to execute command: {:?}", reason))
+    }
+
+    Ok(temp_filename)
+  }
+
   /// Download file from Google Drive into the `TempDir`.
   /// Return the local filesystem filename.
   pub async fn download_file(&self,

crates/lib/mysql_queries/src/payloads/generic_inference_args/image_generation_payload.rs

@@ -74,6 +74,13 @@ pub struct StableDiffusionArgs {
 
     #[serde(rename = "t")]
     pub type_of_inference: String,
+
+    // loRA / checkpoint description
+    #[serde(rename = "de")]
+    pub description: Option<String>,
+    // loRA / checkpoint name and description
+    #[serde(rename = "na")]
+    pub name: Option<String>
 }
 
 

crates/lib/mysql_queries/src/queries/media_files/create/insert_media_file_from_zero_shot_tts.rs

@@ -65,7 +65,7 @@ pub async fn insert_media_file_from_zero_shot(
       maybe_creator_category_synthetic_id = Some(next_lipsync_id);
     }
 
-    // TODO: Ask if it make sense to use ONE function to do the insert 
+    // TODO: Ask if it make sense to use ONE function to do the insert * I agree it does .... use insert media file generic and eventually remove all this.
     // Since fill this out apart of the funtion this seems like the only thing to change.
     const ORIGIN_CATEGORY : MediaFileOriginCategory = MediaFileOriginCategory::Inference;
     const ORIGIN_PRODUCT_CATEGORY : MediaFileOriginProductCategory = MediaFileOriginProductCategory::ZeroShotVoice;

crates/lib/mysql_queries/src/queries/media_files/create/insert_media_file_generic.rs

@@ -0,0 +1,172 @@
+use anyhow::anyhow;
+use sqlx;
+use sqlx::MySqlPool;
+
+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_model_type::MediaFileOriginModelType;
+use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory;
+use enums::by_table::media_files::media_file_type::MediaFileType;
+use errors::AnyhowResult;
+use tokens::tokens::media_files::MediaFileToken;
+use tokens::tokens::weight_model_weight;
+use tokens::tokens::users::UserToken;
+
+use crate::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
+use crate::queries::generic_synthetic_ids::transactional_increment_generic_synthetic_id::transactional_increment_generic_synthetic_id;
+// thought about this it seems like this can be a bit more geneneric instead of having this we can ...
+pub struct InsertArgs<'a> {
+    pub pool: &'a MySqlPool,
+    pub job: &'a AvailableInferenceJob,
+
+    pub maybe_mime_type: Option<&'a str>,
+    pub file_size_bytes: u64,
+    pub sha256_checksum: &'a str,
+
+    pub public_bucket_directory_hash: &'a str,
+    pub maybe_public_bucket_prefix: Option<&'a str>,
+    pub maybe_public_bucket_extension: Option<&'a str>,
+
+    pub is_on_prem: bool,
+
+    pub worker_hostname: &'a str,
+    pub worker_cluster: &'a str,
+
+    pub maybe_creator_user_token: Option<&'a UserToken>,
+    pub maybe_creator_anonymous_visitor_token: Option<&'a AnonymousVisitorTrackingToken>,
+    pub creator_ip_address: &'a str,
+    pub creator_set_visibility: Visibility,
+    pub upload_type: UploadType,
+    pub media_file_type: MediaFileType,
+    pub duration_millis: u64,
+
+    pub media_file_origin: MediaFileOriginProductCategory,
+    pub media_file_origin_product_category: MediaFileOriginProductCategory,
+    pub media_file_origin_model_type: Mediafile,
+    pub maybe_origin_filename: String,
+    pub maybe_origin_model_token:WeightsModelToken
+}
+pub async fn insert_media_file_generic(
+    args: InsertArgs<'_>
+) -> AnyhowResult<(MediaFileToken, u64)>
+{
+
+    let result_token = MediaFileToken::generate();
+
+    let mut maybe_creator_file_synthetic_id : Option<u64> = None;
+    let mut maybe_creator_category_synthetic_id : Option<u64> = None;
+
+    let mut transaction = args.pool.begin().await?;
+
+    // create a user token
+    if let Some(creator_user_token) = args.job.maybe_creator_user_token.as_deref() {
+        let user_token = UserToken::new_from_str(creator_user_token);
+
+        let next_media_file_id = transactional_increment_generic_synthetic_id(
+            &user_token,
+            IdCategory::MediaFile,
+            &mut transaction
+        ).await?;
+
+        let next_lipsync_id = transactional_increment_generic_synthetic_id(
+            &user_token,
+            IdCategory::ZeroShotTtsResult,
+            &mut transaction
+        ).await?;
+
+        maybe_creator_file_synthetic_id = Some(next_media_file_id);
+        maybe_creator_category_synthetic_id = Some(next_lipsync_id);
+    }
+
+    // TODO: Ask if it make sense to use ONE function to do the insert * I agree it does .... use insert media file generic and eventually remove all this.
+    // Since fill this out apart of the funtion this seems like the only thing to change.
+    const ORIGIN_CATEGORY : MediaFileOriginCategory = MediaFileOriginCategory::Inference;
+    const ORIGIN_PRODUCT_CATEGORY : MediaFileOriginProductCategory = MediaFileOriginProductCategory::ZeroShotVoice;
+    const ORIGIN_MODEL_TYPE : MediaFileOriginModelType = MediaFileOriginModelType::VallEX;
+    const MEDIA_TYPE : MediaFileType = MediaFileType::Audio;
+
+    let record_id = {
+        let query_result = sqlx::query!(
+            r#"
+    INSERT INTO media_files
+    SET
+      token = ?,
+
+      origin_category = ?,
+      origin_product_category = ?,
+      maybe_origin_model_type = ?,
+
+      media_type = ?,
+      maybe_mime_type = ?,
+      file_size_bytes = ?,
+
+      checksum_sha2 = ?,
+
+      public_bucket_directory_hash = ?,
+      maybe_public_bucket_prefix = ?,
+      maybe_public_bucket_extension = ?,
+
+      maybe_creator_user_token = ?,
+      maybe_creator_anonymous_visitor_token = ?,
+      creator_ip_address = ?,
+
+      creator_set_visibility = ?,
+
+      maybe_creator_file_synthetic_id = ?,
+      maybe_creator_category_synthetic_id = ?,
+
+      is_generated_on_prem = ?,
+      generated_by_worker = ?,
+      generated_by_cluster = ?
+
+            "#,
+          result_token.as_str(),
+
+          ORIGIN_CATEGORY.to_str(),
+          ORIGIN_PRODUCT_CATEGORY.to_str(),
+          ORIGIN_MODEL_TYPE.to_str(),
+
+          MEDIA_TYPE.to_str(),
+          args.maybe_mime_type,
+          args.file_size_bytes,
+
+          args.sha256_checksum,
+
+          args.public_bucket_directory_hash,
+          args.maybe_public_bucket_prefix,
+          args.maybe_public_bucket_extension,
+
+          args.job.maybe_creator_user_token,
+          args.job.maybe_creator_anonymous_visitor_token,
+          args.job.creator_ip_address,
+
+          args.job.creator_set_visibility.to_str(),
+
+          maybe_creator_file_synthetic_id,
+          maybe_creator_category_synthetic_id,
+
+          args.is_on_prem,
+          args.worker_hostname,
+          args.worker_cluster
+        )
+        .execute(&mut *transaction)
+        .await;
+
+        let record_id = match query_result {
+            Ok(res) => {
+                res.last_insert_id()
+            },
+            Err(err) => {
+                // TODO: handle better
+                //transaction.rollback().await?;
+                return Err(anyhow!("Mysql error: {:?}", err));
+            }
+        };
+
+        record_id
+    };
+
+    transaction.commit().await?;
+
+    Ok((result_token, record_id))
+}

crates/service/job/inference_job/src/job/job_types/image_generation/sd/process_job.rs

@@ -1,8 +1,9 @@
+use std::env::temp_dir;
 use std::ops::Deref;
 use std::path::PathBuf;
 use std::thread;
 use std::thread::Thread;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use cloud_storage::remote_file_manager::remote_cloud_bucket_details::RemoteCloudBucketDetails;
 use mysql_queries::payloads::generic_inference_args::image_generation_payload::StableDiffusionArgs;
 use mysql_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob;
@@ -11,16 +12,36 @@ use crate::job::job_loop::job_success_result::{ JobSuccessResult, ResultEntity }
 use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
 use crate::job_dependencies::JobDependencies;
 use anyhow::anyhow;
+use log::info;
+use sqlx::MySqlPool;
+use cloud_storage::remote_file_manager::media_descriptor::MediaImagePngDescriptor;
 use mysql_queries::payloads::generic_inference_args::generic_inference_args::PolymorphicInferenceArgs;
 
 use cloud_storage::remote_file_manager::remote_cloud_file_manager::RemoteCloudFileClient;
+use cloud_storage::remote_file_manager::weights_descriptor::WeightsLoRADescriptor;
 use enums::by_table::generic_inference_jobs::inference_result_type::InferenceResultType::UploadModel;
+use enums::by_table::generic_synthetic_ids::id_category::IdCategory::MediaFile;
+use enums::by_table::media_files::media_file_type::MediaFileType;
+use enums::by_table::media_uploads::media_upload_source::MediaUploadSource;
+use enums::by_table::media_uploads::media_upload_type::MediaUploadType;
+use enums::by_table::model_weights::weights_category::WeightsCategory;
+use enums::by_table::model_weights::weights_types::WeightsType;
+use enums::common::visibility::Visibility;
 use filesys::path_to_string::path_to_string;
+use filesys::safe_delete_temp_directory::safe_delete_temp_directory;
+use mysql_queries::payloads::media_upload_modification_details::MediaUploadModificationDetails;
+use mysql_queries::queries::media_files::create::insert_media_file_from_file_upload::{insert_media_file_from_file_upload, InsertMediaFileFromUploadArgs, UploadType};
+use mysql_queries::queries::media_uploads::insert_media_upload::{Args, insert_media_upload};
+use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
 use tokens::tokens::model_weights::ModelWeightToken;
+use tokens::tokens::users::UserToken;
 use crate::job::job_types::image_generation::sd::sd_inference_command::InferenceArgs;
 use crate::job::job_types::image_generation::sd::stable_diffusion_dependencies::StableDiffusionDependencies;
 use crate::job::job_types::image_generation::sd::validate_inputs::validate_inputs;
 
+use google_drive_common::google_drive_download_command::GoogleDriveDownloadCommand;
+use mysql_queries::queries::model_weights::create::create_weight::{create_weight, CreateModelWeightsArgs};
+
 pub struct StableDiffusionProcessArgs<'a> {
     pub job_dependencies: &'a JobDependencies,
     pub job: &'a AvailableInferenceJob,
@@ -30,8 +51,6 @@ pub struct StableDiffusionProcessArgs<'a> {
 // run inference
 // insert record into the db with the inference job token complete.
 
-pub async fn download_from_google_link() {}
-
 pub async fn sd_args_from_job(
     args: &StableDiffusionProcessArgs<'_>
 ) -> Result<StableDiffusionArgs, ProcessSingleJobError> {
@@ -87,7 +106,6 @@ pub async fn process_job_checkpoint(args: &StableDiffusionProcessArgs<'_>) -> Re
     //     entity_type: InferenceResultType::UploadModel,
     //     entity_token:
     // };
-
     Ok(JobSuccessResult {
         maybe_result_entity: None,
         inference_duration: Duration::from_secs(0),
@@ -105,6 +123,8 @@ pub async fn process_job_inference(
 ) -> Result<JobSuccessResult, ProcessSingleJobError> {
     let job = args.job;
     let deps = args.job_dependencies;
+    let mysql_pool = &deps.db.mysql_pool;
+
 
     let sd_args = sd_args_from_job(&args).await?;
     let sd_deps = match &args.job_dependencies.job.job_specific_dependencies.maybe_stable_diffusion_dependencies {
@@ -116,6 +136,17 @@ pub async fn process_job_inference(
         }
     };
 
+    let creator_user_token:UserToken;
+    match &job.maybe_creator_user_token {
+        Some(token) => {
+            creator_user_token = UserToken::new_from_str(token);
+        },
+        None => {
+            return Err(ProcessSingleJobError::InvalidJob(anyhow!("Missing Creator User Token")));
+        }
+    }
+
+
     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)))?;
@@ -126,28 +157,66 @@ pub async fn process_job_inference(
     // thread::sleep(seconds) to check the directory
 
     //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
+    let mut 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))?;
 
 
-    let sd_checkpoint_path = work_temp_dir.path().join("sd_checkpoint");
-    let lora_path = work_temp_dir.path().join("lora");
-    let vae_path = work_temp_dir.path().join("vae");
+    let sd_checkpoint_path = work_temp_dir.path().join("sd_checkpoint.safetensors");
+    let lora_path = work_temp_dir.path().join("lora.safetensors");
+    let vae_path = work_temp_dir.path().join("vae.safetensors");
     let output_path = work_temp_dir.path().join("output");
+    let g_drive_path = work_temp_dir.path().join("gdrive");
 
-    println!("Paths to download to:");
-    println!("sd_checkpoint_path:{}",sd_checkpoint_path.display());
-    println!("lora_path:{}",lora_path.display());
-    println!("vae_path:{}",vae_path.display());
-    println!("output_path:{}",output_path.display());
+    info!("Paths to download to:");
+    info!("sd_checkpoint_path:{}",sd_checkpoint_path.display());
+    info!("lora_path:{}",lora_path.display());
+    info!("vae_path:{}",vae_path.display());
+    info!("output_path:{}",output_path.display());
+    info!("tmp_google_drive_path:{}",g_drive_path.display());
+
+    // If it is not inference then it must be loRA or checkpoint downloading
+    if sd_args.type_of_inference != "inference" {
+        let mut download_url= String::from("");
+
+        if sd_args.type_of_inference == "lora" {
+            download_url = match sd_args.maybe_lora_upload_path {
+                Some(val) => { val },
+                None => { "".to_string() }
+            };
+        }
+        else if sd_args.type_of_inference == "checkpoint" {
+            download_url = match sd_args.maybe_upload_path {
+                Some(val) => { val },
+                None => { "".to_string() }
+            }
+        }
+
+        if download_url.len() <= 0 {
+            return Err(ProcessSingleJobError::from_anyhow_error(anyhow!("Failed to Download URL Missing")));
+        }
+        let file_name = "download.safetensors";
+        let download_script = easyenv::get_env_string_or_default(
+            "DOWNLOAD_SCRIPT",
+            "./scripts/download_internet_file.py");
+        let google_drive_downloader =
+            GoogleDriveDownloadCommand::new_production(&download_script);
+        info!("Downloading {}",download_url);
+        let download_filename = match google_drive_downloader.download_file_with_file_name(&download_url, &work_temp_dir,file_name).await {
+            Ok(filename) => filename,
+            Err(e) => {
+                return Err(ProcessSingleJobError::from_anyhow_error(anyhow!("Failed to Download")));
+            }
+        };
+        let download_file_path = work_temp_dir.path().join(file_name);
+        info!("File Retrieved at {}",download_file_path.display());
+    }
 
     // // Unpack loRA and Checkpoint
     // // run inference by downloading from google drive.
     let lora_token = sd_args.maybe_lora_model_token;
     let weight_token = sd_args.maybe_sd_model_token;
 
-
     let retrieved_sd_record = match weight_token {
         Some(token) => {
             let retrieved_sd_record = get_weight_by_token(
@@ -272,11 +341,11 @@ pub async fn process_job_inference(
             20
         }
     };
+    let inference_start_time = Instant::now();
 
-    // #TODO allow for batch and saving later. output files
     sd_deps.inference_command.execute_inference(InferenceArgs {
         work_dir: work_temp_dir.path().to_path_buf(),
-        output_file: output_path,
+        output_file: output_path.clone(),
         stderr_output_file: stderr_output_file,
         prompt: prompt,
         negative_prompt:sd_args.maybe_n_prompt.unwrap_or_default(),
@@ -292,15 +361,95 @@ pub async fn process_job_inference(
         batch_count: sd_args.maybe_batch_count.unwrap_or(1),
     });
 
-    // hack to check the directory before clean up.
-    let thirtyMinutes = 1800;
-    thread::sleep(Duration::from_secs(thirtyMinutes));
-    // upload media and create a record.
+    let inference_duration = Instant::now().duration_since(inference_start_time);
+    let creator_ip_address = &job.creator_ip_address;
+    // run a for loop for output images output_0 in the folder then use upload media.
+    // pngs
+
+    if sd_args.type_of_inference == "inference" {
+        // gather all the images into a single batch
+
+
+        for i in 0..sd_args.maybe_batch_count.unwrap_or(1) {
+            let path = output_path.clone();
+            let file_path = format!("{}_{}.png", path_to_string(path), i);
+            println!("Upload File Path:{}", file_path);
+            let metadata = remote_cloud_file_client.upload_file(Box::new(MediaImagePngDescriptor {}), file_path.as_ref()).await?;
+
+            let bucket_details = match metadata.bucket_details {
+                Some(val) => { val }
+                None => return Err(ProcessSingleJobError::from_anyhow_error(anyhow!("no VAE? thats a problem.")))
+            };
+
+            // TODO: Generic Insert with batch id.
+            // insert media file generic should run here.
+            insert_media_file_from_file_upload(InsertMediaFileFromUploadArgs {
+                pool: mysql_pool,
+                maybe_creator_user_token: Some(&creator_user_token),
+                maybe_creator_anonymous_visitor_token: None,
+                creator_ip_address: creator_ip_address,
+                creator_set_visibility: Default::default(),
+                upload_type: UploadType::Filesystem,
+                media_file_type: MediaFileType::Image,
+                maybe_mime_type: Some(metadata.mimetype.as_ref()),
+                file_size_bytes: metadata.file_size_bytes,
+                duration_millis: inference_duration.as_secs() * 1000,
+                sha256_checksum: metadata.sha256_checksum.as_str(),
+                public_bucket_directory_hash: bucket_details.object_hash.as_str(),
+                maybe_public_bucket_prefix: Some(bucket_details.prefix.as_str()),
+                maybe_public_bucket_extension: Some(bucket_details.suffix.as_str()),
+            }).await?;
+        }
+
+        // hack to check the directory before clean up.
+        let thirtyMinutes = 1800;
+        thread::sleep(Duration::from_secs(thirtyMinutes));
+        // upload media and create a record.
+
+        Ok(JobSuccessResult {// TODO: batch token needs to go here
+            maybe_result_entity: None,
+            inference_duration: Duration::from_secs(0),
+        })
+    }
+    else { // must be an upload and need upload token
+
+
+        // The parameters will be updated on another screen perhaps?
+        // so right now it will fill with the availible  values.
+        create_weight(CreateModelWeightsArgs {
+            token: &ModelWeightToken::generate(),
+            weights_type: WeightsType::StableDiffusion15,
+            weights_category: WeightsCategory::ImageGeneration,
+            title: "".to_string(),
+            maybe_description_markdown: None,
+            maybe_description_rendered_html: None,
+            creator_user_token: Some(&creator_user_token),
+            creator_ip_address: creator_ip_address,
+            creator_set_visibility: Default::default(),
+            maybe_last_update_user_token: None,
+            original_download_url: None,
+            original_filename: None,
+            file_size_bytes: 0,
+            file_checksum_sha2: "".to_string(),
+            public_bucket_hash: "".to_string(),
+            maybe_public_bucket_prefix: None,
+            maybe_public_bucket_extension: None,
+            cached_user_ratings_total_count: 0,
+            cached_user_ratings_positive_count: 0,
+            cached_user_ratings_negative_count: 0,
+            maybe_cached_user_ratings_ratio: None,
+            version: 0,
+            mysql_pool,
+        }).await?;
+
+        Ok(JobSuccessResult {// TODO: batch token needs to go here
+            maybe_result_entity: None,
+            inference_duration: Duration::from_secs(0),
+        })
+    }
+
+
 
-    Ok(JobSuccessResult {
-        maybe_result_entity: None,
-        inference_duration: Duration::from_secs(0),
-    })
 }
 
 #[cfg(test)]