-

package org.hammerlab.guacamole.commands

import org.apache.spark.SparkContext
import org.bdgenomics.formats.avro.AlignmentRecord
import org.hammerlab.guacamole.{ DistributedUtil, SparkCommand, Common }
import org.bdgenomics.adam.rdd.ADAMContext._
import org.kohsuke.args4j.Option
import org.bdgenomics.adam.rdd.ADAMSaveAnyArgs

/**
 * Created by danvk on 6/25/15.
 */
object StructuralVariant {

  protected class Arguments extends Common.Arguments.Reads with DistributedUtil.Arguments with ADAMSaveAnyArgs {
    @Option(name = "--filter-contig", usage = "Filter to alignments where either mate is in this contig.")
    var filterContig: String = ""

    // These are required by ADAMSaveAnyArgs
    @Option(name = "--out", required = true, usage = "Location to write the transformed data in ADAM/Parquet format")
    var outputPath: String = null
    @Option(required = false, name = "-sort_fastq_output", usage = "Sets whether to sort the FASTQ output, if saving as FASTQ. False by default. Ignored if not saving as FASTQ.")
    var sortFastqOutput: Boolean = false
  }

  object Caller extends SparkCommand[Arguments] {
    override val name = "structural-variant"
    override val description = "TBD"

    def matchesContig(alignment: AlignmentRecord, contig: String) = {
      alignment.getContig == contig || (alignment.getMateContig != null && alignment.getMateContig.getContigName == contig)
    }

    override def run(args: Arguments, sc: SparkContext): Unit = {

      val filterContig = args.filterContig

      val alignments = sc.loadAlignments(args.reads)

      val matchingAlignments = alignments.filter(matchesContig(_, filterContig))
      matchingAlignments.persist()

      println("Found " + matchingAlignments.count() + " alignments with   one pair in " + filterContig)

      matchingAlignments.coalesce(10).adamSAMSave(args.outputPath, asSam = false)

    }
  }

}