Removed run_many! from deep_somatic;rs

src/callers/deep_somatic.rs

@@ -95,27 +95,15 @@ use rust_htslib::bam::{self, Read};
 use uuid::Uuid;
 
 use crate::{
-    annotation::{Annotation, Annotations, Caller, CallerCat, Sample},
-    collection::vcf::Vcf,
-    commands::{
-        bcftools::{BcftoolsConcat, BcftoolsIndex, BcftoolsKeepPass},
-        Command as JobCommand, LocalBatchRunner, LocalRunner, SbatchRunner, SlurmParams,
-        SlurmRunner,
-    },
-    config::Config,
-    helpers::{
+    annotation::{Annotation, Annotations, Caller, CallerCat, Sample}, collection::vcf::Vcf, commands::{
+        AnyJob, Command as JobCommand, LocalBatchRunner, LocalRunner, SbatchRunner, SlurmParams, SlurmRunner, bcftools::{BcftoolsConcat, BcftoolsIndex, BcftoolsKeepPass}, exec_jobs
+    }, config::Config, helpers::{
         get_genome_sizes_in_header_order, is_file_older, remove_dir_if_exists,
         singularity_bind_flags, split_ordered_genome_into_n_regions_exact,
-    },
-    io::vcf::read_vcf,
-    locker::SampleLock,
-    pipes::{Initialize, ShouldRun, Version},
-    run, run_many,
-    runners::Run,
-    variant::{
+    }, io::vcf::read_vcf, jobs_seq, locker::SampleLock, pipes::{Initialize, ShouldRun, Version}, run, runners::Run, variant::{
         variant_collection::VariantCollection,
         vcf_variant::{Label, Variants},
-    },
+    }
 };
 
 /// DeepSomatic paired tumor-normal somatic variant caller.
@@ -551,9 +539,13 @@ fn merge_deepsomatic_parts(base: &DeepSomatic, n_parts: usize) -> anyhow::Result
         n_parts, final_passed_vcf
     );
 
-    let mut concat = BcftoolsConcat::from_config(&base.config, part_pass_paths, &final_tmp);
-    run!(&base.config, &mut concat)
-        .context("Failed to run bcftools concat for DeepSomatic parts")?;
+    let concat = BcftoolsConcat::from_config(&base.config, part_pass_paths, &final_tmp);
+
+    let _ = exec_jobs(
+        vec![jobs_seq![concat]],
+        base.config.slurm_runner,
+        base.config.slurm_max_par.into(),
+    ).context("Failed to run bcftools concat for DeepSomatic parts")?;
 
     fs::rename(&final_tmp, &final_passed_vcf)
         .context("Failed to rename merged DeepSomatic PASS VCF")?;
@@ -605,18 +597,8 @@ fn merge_deepsomatic_parts(base: &DeepSomatic, n_parts: usize) -> anyhow::Result
 pub fn run_deepsomatic_chunked(id: &str, config: &Config, n_parts: usize) -> anyhow::Result<()> {
     anyhow::ensure!(n_parts > 0, "n_parts must be > 0");
 
-    // Lock at the chunked level (caller may have already locked, but this is idempotent protection)
-    // let lock_dir = format!("{}/locks", config.result_dir);
-    // let _lock = SampleLock::acquire(&lock_dir, id, "deepsomatic")
-    //     .with_context(|| format!("Cannot start DeepSomatic chunked for {}", id))?;
-
     let base = DeepSomatic::initialize(id, config)?;
 
-    // if !base.should_run() {
-    //     debug!("DeepSomatic PASS VCF already up-to-date for {id}, skipping.");
-    //     return Ok(());
-    // }
-
     // Get genome sizes from tumor BAM header
     let tumor_bam = config.tumoral_bam(id);
     let reader = bam::Reader::from_path(&tumor_bam)
@@ -635,34 +617,60 @@ pub fn run_deepsomatic_chunked(id: &str, config: &Config, n_parts: usize) -> any
         actual_n_parts, id
     );
 
-    let mut jobs = Vec::with_capacity(actual_n_parts);
-    for (i, regions) in region_chunks.into_iter().enumerate() {
-        let mut job = base.clone();
-        job.regions = regions;
-        job.part_index = Some(i + 1);
-        job.log_dir = format!("{}/part{}", base.log_dir, i + 1);
-        info!("Planned DeepSomatic job:\n{job}");
-        jobs.push(job);
-    }
+    let groups: anyhow::Result<Vec<Vec<Box<dyn AnyJob>>>> = region_chunks
+        .into_iter()
+        .enumerate()
+        .map(|(i, regions)| {
+            let mut job = base.clone();
+            job.regions = regions;
+            job.part_index = Some(i + 1);
+            job.log_dir = format!("{}/part{}", base.log_dir, i + 1);
+            info!("Planned DeepSomatic job:\n{job}");
 
-    // Run DeepSomatic jobs
-    let outputs = run_many!(config, jobs.clone())?;
-    for output in outputs.iter() {
-        output.save_to_file(format!("{}/deepsomatic_", base.log_dir))?;
-    }
+            let pass = BcftoolsKeepPass::from_config(&job.config, job.output_vcf_path(), job.passed_vcf_path());
 
-    // Filter PASS variants for each part in parallel
-    info!(
-        "Filtering PASS variants for all {} parts in parallel",
-        actual_n_parts
-    );
-    let filter_jobs: Vec<_> = jobs
-        .iter()
-        .map(|job| {
-            BcftoolsKeepPass::from_config(&job.config, job.output_vcf_path(), job.passed_vcf_path())
+            Ok(jobs_seq![job, pass])
         })
         .collect();
-    run_many!(config, filter_jobs)?;
+
+    let outputs = exec_jobs(
+        groups?,
+        base.config.slurm_runner,
+        base.config.slurm_max_par.into(),
+    )?;
+
+    for output in &outputs {
+        output.save_to_file(format!("{}/deep_somatic_", base.log_dir))?;
+    }
+
+    // let mut jobs = Vec::with_capacity(actual_n_parts);
+    // for (i, regions) in region_chunks.into_iter().enumerate() {
+    //     let mut job = base.clone();
+    //     job.regions = regions;
+    //     job.part_index = Some(i + 1);
+    //     job.log_dir = format!("{}/part{}", base.log_dir, i + 1);
+    //     info!("Planned DeepSomatic job:\n{job}");
+    //     jobs.push(job);
+    // }
+    //
+    // // Run DeepSomatic jobs
+    // let outputs = run_many!(config, jobs.clone())?;
+    // for output in outputs.iter() {
+    //     output.save_to_file(format!("{}/deepsomatic_", base.log_dir))?;
+    // }
+
+    // Filter PASS variants for each part in parallel
+    // info!(
+    //     "Filtering PASS variants for all {} parts in parallel",
+    //     actual_n_parts
+    // );
+    // let filter_jobs: Vec<_> = jobs
+    //     .iter()
+    //     .map(|job| {
+    //         BcftoolsKeepPass::from_config(&job.config, job.output_vcf_path(), job.passed_vcf_path())
+    //     })
+    //     .collect();
+    // run_many!(config, filter_jobs)?;
 
     // Merge PASS VCFs
     merge_deepsomatic_parts(&base, actual_n_parts)?;