Merge branch 'feature-centrifuge_pipe' into 'develop'

Feature centrifuge pipe



See merge request !471

gears/src/main/scala/nl/lumc/sasc/biopet/pipelines/gears/GearsCentrifuge.scala

 package nl.lumc.sasc.biopet.pipelines.gears
 
-import nl.lumc.sasc.biopet.core.SampleLibraryTag
+import nl.lumc.sasc.biopet.core.{ BiopetFifoPipe, SampleLibraryTag }
 import nl.lumc.sasc.biopet.core.summary.SummaryQScript
-import nl.lumc.sasc.biopet.extensions.Gzip
+import nl.lumc.sasc.biopet.extensions.{ Gzip, Zcat }
 import nl.lumc.sasc.biopet.extensions.centrifuge.{ Centrifuge, CentrifugeKreport }
 import nl.lumc.sasc.biopet.extensions.tools.KrakenReportToJson
 import nl.lumc.sasc.biopet.utils.config.Configurable
@@ -27,30 +27,31 @@ class GearsCentrifuge(val root: Configurable) extends QScript with SummaryQScrip
   }
 
   def centrifugeOutput = new File(outputDir, s"$outputName.centrifuge.gz")
+  def centrifugeMetOutput = new File(outputDir, s"$outputName.centrifuge.met")
 
   def biopetScript(): Unit = {
     val centrifuge = new Centrifuge(this)
     centrifuge.inputR1 = fastqR1
     centrifuge.inputR2 = fastqR2
-    centrifuge.output = new File(outputDir, s"$outputName.centrifuge")
     centrifuge.report = Some(new File(outputDir, s"$outputName.centrifuge.report"))
-    centrifuge.isIntermediate = true
-    add(centrifuge)
+    centrifuge.metFile = Some(centrifugeMetOutput)
+    val centrifugeCmd = centrifuge | new Gzip(this) > centrifugeOutput
+    centrifugeCmd.threadsCorrection = -1
+    add(centrifugeCmd)
 
-    add(Gzip(this, centrifuge.output, centrifugeOutput))
-
-    makeKreport(List(centrifuge.output), "centrifuge", unique = false)
-    makeKreport(List(centrifuge.output), "centrifuge_unique", unique = true)
+    makeKreport("centrifuge", unique = false)
+    makeKreport("centrifuge_unique", unique = true)
 
     addSummaryJobs()
   }
 
-  protected def makeKreport(inputFiles: List[File], name: String, unique: Boolean): Unit = {
+  protected def makeKreport(name: String, unique: Boolean): Unit = {
+    val fifo = new File(outputDir, s"$outputName.$name.fifo")
     val centrifugeKreport = new CentrifugeKreport(this)
-    centrifugeKreport.centrifugeOutputFiles = inputFiles
+    centrifugeKreport.centrifugeOutputFiles :+= fifo
     centrifugeKreport.output = new File(outputDir, s"$outputName.$name.kreport")
     centrifugeKreport.onlyUnique = unique
-    add(centrifugeKreport)
+    add(new BiopetFifoPipe(this, List(centrifugeKreport, Zcat(this, centrifugeOutput, fifo))))
 
     val krakenReportJSON = new KrakenReportToJson(this)
     krakenReportJSON.inputReport = centrifugeKreport.output
@@ -72,5 +73,4 @@ class GearsCentrifuge(val root: Configurable) extends QScript with SummaryQScrip
       case Some(file) => Map("input_R2" -> file)
       case _          => Map()
     })
-
 }

gears/src/test/scala/nl/lumc/sasc/biopet/pipelines/gears/GearsSingleTest.scala

@@ -17,7 +17,7 @@ package nl.lumc.sasc.biopet.pipelines.gears
 import java.io.File
 
 import com.google.common.io.Files
-import nl.lumc.sasc.biopet.core.BiopetPipe
+import nl.lumc.sasc.biopet.core.BiopetCommandLineFunction
 import nl.lumc.sasc.biopet.extensions.centrifuge.{ Centrifuge, CentrifugeKreport }
 import nl.lumc.sasc.biopet.extensions.kraken.{ Kraken, KrakenReport }
 import nl.lumc.sasc.biopet.extensions.picard.SamToFastq
@@ -106,7 +106,8 @@ abstract class TestGearsSingle extends TestNGSuite with Matchers {
         gears.outputName shouldBe (if (inputMode == Some("bam")) "bamfile" else "R1")
       }
 
-      val pipesJobs = gears.functions.filter(_.isInstanceOf[BiopetPipe]).flatMap(_.asInstanceOf[BiopetPipe].pipesJobs)
+      val pipesJobs = gears.functions.filter(_.isInstanceOf[BiopetCommandLineFunction])
+        .flatMap(_.asInstanceOf[BiopetCommandLineFunction].pipesJobs)
 
       gears.summarySettings("gears_use_kraken") shouldBe kraken.getOrElse(false)
       gears.summarySettings("gear_use_qiime_rtax") shouldBe qiimeRtax
@@ -129,8 +130,8 @@ abstract class TestGearsSingle extends TestNGSuite with Matchers {
       gears.functions.count(_.isInstanceOf[KrakenReportToJson]) shouldBe
         ((if (kraken.getOrElse(false)) 1 else 0) + (if (centrifuge) 2 else 0))
 
-      gears.functions.count(_.isInstanceOf[Centrifuge]) shouldBe (if (centrifuge) 1 else 0)
-      gears.functions.count(_.isInstanceOf[CentrifugeKreport]) shouldBe (if (centrifuge) 2 else 0)
+      pipesJobs.count(_.isInstanceOf[Centrifuge]) shouldBe (if (centrifuge) 1 else 0)
+      pipesJobs.count(_.isInstanceOf[CentrifugeKreport]) shouldBe (if (centrifuge) 2 else 0)
     }
   }
 }

mapping/src/test/scala/nl/lumc/sasc/biopet/pipelines/mapping/MappingTest.scala

@@ -17,8 +17,8 @@ package nl.lumc.sasc.biopet.pipelines.mapping
 import java.io.{File, FileOutputStream}
 
 import com.google.common.io.Files
+import nl.lumc.sasc.biopet.core.BiopetCommandLineFunction
 import nl.lumc.sasc.biopet.extensions.centrifuge.Centrifuge
-import nl.lumc.sasc.biopet.extensions.kraken.Kraken
 import nl.lumc.sasc.biopet.pipelines.flexiprep.Fastqc
 import nl.lumc.sasc.biopet.utils.ConfigUtils
 import nl.lumc.sasc.biopet.utils.config.Config
@@ -26,7 +26,7 @@ import org.apache.commons.io.FileUtils
 import org.broadinstitute.gatk.queue.QSettings
 import org.scalatest.Matchers
 import org.scalatest.testng.TestNGSuite
-import org.testng.annotations.{AfterClass, BeforeClass, DataProvider, Test}
+import org.testng.annotations.{ AfterClass, BeforeClass, DataProvider, Test }
 
 /**
  * Test class for [[Mapping]]
@@ -86,10 +86,13 @@ abstract class AbstractTestMapping(val aligner: String) extends TestNGSuite with
     mapping.libId = Some("1")
     mapping.script()
 
+    val pipesJobs = mapping.functions.filter(_.isInstanceOf[BiopetCommandLineFunction])
+      .flatMap(_.asInstanceOf[BiopetCommandLineFunction].pipesJobs)
+
     //Flexiprep
     mapping.functions.count(_.isInstanceOf[Fastqc]) shouldBe (if (skipFlexiprep) 0 else if (paired) 4 else 2)
 
-    mapping.functions.count(_.isInstanceOf[Centrifuge]) shouldBe (if (unmappedToGears) 1 else 0)
+    pipesJobs.count(_.isInstanceOf[Centrifuge]) shouldBe (if (unmappedToGears) 1 else 0)
   }
 
   val outputDir = Files.createTempDir()

mapping/src/test/scala/nl/lumc/sasc/biopet/pipelines/mapping/MultisampleMappingTest.scala

@@ -14,18 +14,18 @@
  */
 package nl.lumc.sasc.biopet.pipelines.mapping
 
-import java.io.{File, FileOutputStream}
+import java.io.{ File, FileOutputStream }
 
 import com.google.common.io.Files
+import nl.lumc.sasc.biopet.core.BiopetCommandLineFunction
 import nl.lumc.sasc.biopet.extensions.centrifuge.Centrifuge
-import nl.lumc.sasc.biopet.extensions.kraken.Kraken
-import nl.lumc.sasc.biopet.extensions.picard.{MarkDuplicates, MergeSamFiles}
-import nl.lumc.sasc.biopet.utils.{ConfigUtils, Logging}
+import nl.lumc.sasc.biopet.extensions.picard.{ MarkDuplicates, MergeSamFiles }
+import nl.lumc.sasc.biopet.utils.{ ConfigUtils, Logging }
 import nl.lumc.sasc.biopet.utils.config.Config
 import org.broadinstitute.gatk.queue.QSettings
 import org.scalatest.Matchers
 import org.scalatest.testng.TestNGSuite
-import org.testng.annotations.{DataProvider, Test}
+import org.testng.annotations.{ DataProvider, Test }
 
 /**
  * Created by pjvanthof on 15/05/16.
@@ -88,6 +88,9 @@ trait MultisampleMappingTestTrait extends TestNGSuite with Matchers {
       val numberFastqLibs = (if (sample1) 1 else 0) + (if (sample2) 2 else 0) + (if (sample3 && bamToFastq) 1 else 0) + (if (sample4 && bamToFastq) 1 else 0)
       val numberSamples = (if (sample1) 1 else 0) + (if (sample2) 1 else 0)
 
+      val pipesJobs = pipeline.functions.filter(_.isInstanceOf[BiopetCommandLineFunction])
+        .flatMap(_.asInstanceOf[BiopetCommandLineFunction].pipesJobs)
+
       import MultisampleMapping.MergeStrategy
       pipeline.functions.count(_.isInstanceOf[MarkDuplicates]) shouldBe (numberFastqLibs +
         (if (sample2 && (merge == MergeStrategy.MarkDuplicates || merge == MergeStrategy.PreProcessMarkDuplicates)) 1 else 0))
@@ -103,7 +106,7 @@ trait MultisampleMappingTestTrait extends TestNGSuite with Matchers {
           }
       }
 
-      pipeline.functions.count(_.isInstanceOf[Centrifuge]) shouldBe (if (unmappedToGears) (numberFastqLibs + numberSamples) else 0)
+      pipesJobs.count(_.isInstanceOf[Centrifuge]) shouldBe (if (unmappedToGears) (numberFastqLibs + numberSamples) else 0)
 
       pipeline.summarySettings.get("merge_strategy") shouldBe Some(merge.toString)
     }