Implementation of BySampleSAMFileWriter

ReduceReads now works with the n-way-out capability, splitting by sample.
DEV-27 #resolve #time 3m

protected/java/src/org/broadinstitute/sting/gatk/walkers/compression/reducereads/ReduceReads.java

@@ -25,6 +25,9 @@
 
 package org.broadinstitute.sting.gatk.walkers.compression.reducereads;
 
+import net.sf.samtools.SAMFileHeader;
+import net.sf.samtools.SAMFileWriter;
+import net.sf.samtools.SAMProgramRecord;
 import net.sf.samtools.util.SequenceUtil;
 import org.broadinstitute.sting.commandline.Argument;
 import org.broadinstitute.sting.commandline.Hidden;
@@ -45,6 +48,7 @@ import org.broadinstitute.sting.utils.Utils;
 import org.broadinstitute.sting.utils.clipping.ReadClipper;
 import org.broadinstitute.sting.utils.exceptions.ReviewedStingException;
 import org.broadinstitute.sting.utils.help.DocumentedGATKFeature;
+import org.broadinstitute.sting.utils.sam.BySampleSAMFileWriter;
 import org.broadinstitute.sting.utils.sam.GATKSAMRecord;
 import org.broadinstitute.sting.utils.sam.ReadUtils;
 
@@ -86,7 +90,8 @@ import java.util.*;
 public class ReduceReads extends ReadWalker<LinkedList<GATKSAMRecord>, ReduceReadsStash> {
 
     @Output
-    private StingSAMFileWriter out;
+    private StingSAMFileWriter out = null;
+    private SAMFileWriter writerToUse = null;
 
     /**
      * The number of bases to keep around mismatches (potential variation)
@@ -196,6 +201,10 @@ public class ReduceReads extends ReadWalker<LinkedList<GATKSAMRecord>, ReduceRea
     @Argument(fullName = "contigs", shortName = "ctg", doc = "", required = false)
     private int nContigs = 2;
 
+    @Hidden
+    @Argument(fullName = "nwayout", shortName = "nw", doc = "", required = false)
+    private boolean nwayout = false;
+
     @Hidden
     @Argument(fullName = "", shortName = "dl", doc = "", required = false)
     private int debugLevel = 0;
@@ -227,6 +236,7 @@ public class ReduceReads extends ReadWalker<LinkedList<GATKSAMRecord>, ReduceRea
     SortedSet<GenomeLoc> intervalList;
     
     private static final String PROGRAM_RECORD_NAME = "GATK ReduceReads";   // The name that will go in the @PG tag
+    private static final String PROGRAM_FILENAME_EXTENSION = ".reduced.bam";
 
     /**
      * Basic generic initialization of the readNameHash and the intervalList. Output initialization
@@ -242,10 +252,24 @@ public class ReduceReads extends ReadWalker<LinkedList<GATKSAMRecord>, ReduceRea
         if (toolkit.getIntervals() != null)
             intervalList.addAll(toolkit.getIntervals());
 
-        if (!NO_PG_TAG)
+
-            Utils.setupWriter(out, toolkit, false, true, this, PROGRAM_RECORD_NAME);
+        // todo -- rework the whole NO_PG_TAG thing
-        else
+        final boolean preSorted = true;
+        final boolean indexOnTheFly = true;
+        final boolean generateMD5 = true;
+        final boolean keep_records = true;
+        final SAMFileHeader.SortOrder sortOrder = SAMFileHeader.SortOrder.coordinate;
+        if (nwayout) {
+            SAMProgramRecord programRecord = NO_PG_TAG ? null : Utils.createProgramRecord(toolkit, this, PROGRAM_RECORD_NAME);
+            writerToUse = new BySampleSAMFileWriter(toolkit, PROGRAM_FILENAME_EXTENSION, sortOrder, preSorted, indexOnTheFly, NO_PG_TAG, programRecord, true);
+        }
+        else {
+            writerToUse = out;
             out.setPresorted(false);
+            if (!NO_PG_TAG) {
+                Utils.setupWriter(out, toolkit, !preSorted, keep_records, this, PROGRAM_RECORD_NAME);
+            }
+        }
     }
 
     /**
@@ -386,6 +410,9 @@ public class ReduceReads extends ReadWalker<LinkedList<GATKSAMRecord>, ReduceRea
         // output any remaining reads in the compressor
         for (GATKSAMRecord read : stash.close())
             outputRead(read);
+
+        if (nwayout)
+            writerToUse.close();
     }
 
     /**
@@ -554,7 +581,7 @@ public class ReduceReads extends ReadWalker<LinkedList<GATKSAMRecord>, ReduceRea
         if (!DONT_COMPRESS_READ_NAMES)
             compressReadName(read);
 
-        out.addAlignment(read);
+        writerToUse.addAlignment(read);
     }
 
     /**

public/java/src/org/broadinstitute/sting/utils/sam/BySampleSAMFileWriter.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2010 The Broad Institute
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following
+ * conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+ * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+ * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+ * OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package org.broadinstitute.sting.utils.sam;
+
+import net.sf.samtools.SAMFileHeader;
+import net.sf.samtools.SAMProgramRecord;
+import net.sf.samtools.SAMReadGroupRecord;
+import net.sf.samtools.SAMRecord;
+import org.broadinstitute.sting.gatk.GenomeAnalysisEngine;
+import org.broadinstitute.sting.gatk.datasources.reads.SAMReaderID;
+import org.broadinstitute.sting.utils.exceptions.ReviewedStingException;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Created by IntelliJ IDEA.
+ * User: carneiro
+ * Date: Nov 13
+ */
+public class BySampleSAMFileWriter extends NWaySAMFileWriter{
+
+    private final Map<String, SAMReaderID> sampleToWriterMap;
+
+    public BySampleSAMFileWriter(GenomeAnalysisEngine toolkit, String ext, SAMFileHeader.SortOrder order, boolean presorted, boolean indexOnTheFly, boolean generateMD5, SAMProgramRecord pRecord, boolean keep_records) {
+        super(toolkit, ext, order, presorted, indexOnTheFly, generateMD5, pRecord, keep_records);
+
+        sampleToWriterMap = new HashMap<String, SAMReaderID>(toolkit.getSAMFileHeader().getReadGroups().size() * 2);
+
+        for (SAMReaderID readerID : toolkit.getReadsDataSource().getReaderIDs()) {
+            for (SAMReadGroupRecord rg : toolkit.getReadsDataSource().getHeader(readerID).getReadGroups()) {
+                String sample = rg.getSample();
+                if (sampleToWriterMap.containsKey(sample) && sampleToWriterMap.get(sample) != readerID) {
+                    throw new ReviewedStingException("The same sample appears in multiple files, this input cannot be multiplexed using the BySampleSAMFileWriter, try NWaySAMFileWriter instead.");
+                }
+                else {
+                    sampleToWriterMap.put(sample, readerID);
+                }
+            }
+        }
+    }
+
+    @Override
+    public void addAlignment(SAMRecord samRecord) {
+        super.addAlignment(samRecord, sampleToWriterMap.get(samRecord.getReadGroup().getSample()));
+    }
+}

public/java/src/org/broadinstitute/sting/utils/sam/NWaySAMFileWriter.java

@@ -28,10 +28,8 @@ package org.broadinstitute.sting.utils.sam;
 import net.sf.samtools.*;
 import org.broadinstitute.sting.gatk.GenomeAnalysisEngine;
 import org.broadinstitute.sting.gatk.datasources.reads.SAMReaderID;
-import org.broadinstitute.sting.gatk.io.StingSAMFileWriter;
 import org.broadinstitute.sting.utils.exceptions.StingException;
 import org.broadinstitute.sting.utils.exceptions.UserException;
-import org.broadinstitute.sting.utils.text.TextFormattingUtils;
 
 import java.io.File;
 import java.util.*;
@@ -173,7 +171,11 @@ public class NWaySAMFileWriter implements SAMFileWriter {
             String rg_orig = toolkit.getReadsDataSource().getOriginalReadGroupId(rg);
             samRecord.setAttribute("RG",rg_orig);
         }
-        writerMap.get(id).addAlignment(samRecord);
+        addAlignment(samRecord, id);
+    }
+
+    public void addAlignment(SAMRecord samRecord, SAMReaderID readerID) {
+        writerMap.get(readerID).addAlignment(samRecord);
     }
 
     public SAMFileHeader getFileHeader() {