Using picard for bam merging.

Properties now propagate to scatter/gather functions.

git-svn-id: file:///humgen/gsa-scr1/gsa-engineering/svn_contents/trunk@3659 348d0f76-0448-11de-a6fe-93d51630548a

scala/src/org/broadinstitute/sting/queue/QArguments.scala

@@ -12,7 +12,7 @@ class QArguments(args: Array[String]) {
   var dryRun = false
   val scripts = new ListBuffer[String]
   var inputPaths = List.empty[File]
-  var argMap = Map.empty[String, String]
+  var properties = Map.empty[String, String]
 
   val userArgs = parseArgs(args)
 
@@ -29,7 +29,7 @@ class QArguments(args: Array[String]) {
     if (isFlagged(filtered, "-bsub"))
       bsubAllJobs = true
     for (arg <- getArgs(filtered, "-P"))
-      addArg(arg)
+      addProperties(arg)
     for (arg <- getArgs(filtered, "-I"))
       addFile(arg)
     for (arg <- getArgs(filtered, "-S"))
@@ -64,18 +64,18 @@ class QArguments(args: Array[String]) {
     found
   }
 
-  def addArg(arg: String) = {
+  def addProperties(arg: String) = {
     var file = new File(arg)
     if (arg.contains("=") && !file.exists) {
       val tokens = arg.split("=", 2)
-      argMap += tokens(0) -> tokens(1)
+      properties += tokens(0) -> tokens(1)
     } else if (arg.endsWith(".properties")) {
       if (!file.exists)
         throw new QException("File not found: " + file.getAbsolutePath)
       var props = new Properties
       props.load(new FileInputStream(file))
       for ((name, value) <- props)
-        argMap += name -> value
+        properties += name -> value
     } else {
       throw new QException("Invalid property: " + arg)
     }

scala/src/org/broadinstitute/sting/queue/QScript.scala

@@ -88,7 +88,7 @@ object QScript {
      * Sets the @Input and @Output values for a single function
      */
     def setParams(function: CommandLineFunction): Unit =
-      for ((name, value) <- qArgs.argMap) function.addOrUpdateWithStringValue(name, value)
+      function.properties = qArgs.properties
 
     /**
      * Executes functions that have been added to the pipeline.

scala/src/org/broadinstitute/sting/queue/function/CommandLineFunction.scala

@@ -6,9 +6,19 @@ import java.lang.annotation.Annotation
 import org.broadinstitute.sting.commandline.{Input, Output, ArgumentDescription}
 
 trait CommandLineFunction extends InputOutputFunction with DispatchFunction {
+  var properties = Map.empty[String, String]
+
   def inputFieldsWithValues = inputFields.filter(hasFieldValue(_))
   def outputFieldsWithValues = outputFields.filter(hasFieldValue(_))
 
+  /**
+   * Sets parameters from the arg map.
+   */
+  override def freeze = {
+    for ((name, value) <- properties) addOrUpdateWithStringValue(name, value)
+    super.freeze
+  }
+
   /**
    * Repeats parameters with a prefix/suffix if they are set otherwise returns "".
    * Skips null, Nil, None.  Unwraps Some(x) to x.  Everything else is called with x.toString.

scala/src/org/broadinstitute/sting/queue/function/InputOutputFunction.scala

@@ -22,10 +22,10 @@ trait InputOutputFunction extends QFunction with Cloneable {
 
   /**
    * Sets a field value using the name of the field.
-   * Field must be annotated with @Input, @Output, or @Internal
+   * Field must be annotated with @Input or @Output
    * @return true if the value was found and set
    */
-  def addOrUpdateWithStringValue(name: String, value: String) = {
+  protected def addOrUpdateWithStringValue(name: String, value: String) = {
     fields.find(_.getName == name) match {
       case Some(field) =>
         val isInput = ReflectionUtils.hasAnnotation(field, classOf[Input])

scala/src/org/broadinstitute/sting/queue/function/scattergather/BamGatherFunction.scala

@@ -1,9 +1,17 @@
 package org.broadinstitute.sting.queue.function.scattergather
 
 import java.io.File
+import org.broadinstitute.sting.commandline.Input
 
 class BamGatherFunction extends GatherFunction {
   type GatherType = File
 
-  def commandLine = "samtools merge %s%s".format(originalOutput, repeat(" ", gatherParts))
+  @Input(doc="Picard MergeSamFiles.jar.  At the Broad this can be found at /seq/software/picard/current/bin/MergeSamFiles.jar.  Outside the broad see http://picard.sourceforge.net/")
+  var picardMergeSamFilesJar: String = _
+
+  @Input(doc="Compression level 1-9", required=false)
+  var picardMergeCompressionLevel: Option[Int] = None
+
+  def commandLine = "java -jar %s%s%s%s".format(picardMergeSamFilesJar,
+    optional(" COMPRESSION_LEVEL=", picardMergeCompressionLevel), " OUTPUT=" + originalOutput, repeat(" INPUT=", gatherParts))
 }

scala/src/org/broadinstitute/sting/queue/function/scattergather/ScatterGatherableFunction.scala

@@ -32,6 +32,7 @@ object ScatterGatherableFunction {
 
     // Create a function that will remove any temporary items
     var cleanupFunction = new CleanupTempDirsFunction
+    cleanupFunction.properties = originalFunction.properties
     cleanupFunction.jobNamePrefix = originalFunction.jobNamePrefix
     cleanupFunction.commandDirectory = originalFunction.commandDirectory
 
@@ -42,6 +43,7 @@ object ScatterGatherableFunction {
     // Create the scatter function based on @Scatter
     val scatterFunction = getScatterFunction(scatterField)
     scatterFunction.setOriginalFunction(originalFunction)
+    scatterFunction.properties = originalFunction.properties
     scatterFunction.jobNamePrefix = originalFunction.jobNamePrefix
     scatterFunction.commandDirectory = originalFunction.temp("scatter-" + scatterField.getName)
     scatterFunction.originalInput = originalValue.asInstanceOf[scatterFunction.ScatterType]
@@ -55,6 +57,7 @@ object ScatterGatherableFunction {
       // Create the gather function based on @Gather
       val gatherFunction = getGatherFunction(outputField)
       gatherFunction.setOriginalFunction(originalFunction)
+      gatherFunction.properties = originalFunction.properties
       gatherFunction.jobNamePrefix = originalFunction.jobNamePrefix
       gatherFunction.commandDirectory = originalFunction.temp("gather-" + outputField.getName)
 
@@ -96,6 +99,7 @@ object ScatterGatherableFunction {
     // Create a function to create all of the temp directories.
     // All of its inputs are the inputs of the original function.
     val initializeFunction = new CreateTempDirsFunction
+    initializeFunction.properties = originalFunction.properties
     initializeFunction.jobNamePrefix = originalFunction.jobNamePrefix
     initializeFunction.commandDirectory = originalFunction.commandDirectory