From 5066b143355319dcb5fb0a4ae39a0c0d539e6d8a Mon Sep 17 00:00:00 2001
From: Mark DePristo <depristo@broadinstitute.org>
Date: Sat, 25 Aug 2012 17:19:57 -0400
Subject: [PATCH] Parallel FlagStat

---
 .../sting/gatk/walkers/FlagStat.java          | 111 +++++++++++-------
 1 file changed, 69 insertions(+), 42 deletions(-)

diff --git a/public/java/src/org/broadinstitute/sting/gatk/walkers/FlagStat.java b/public/java/src/org/broadinstitute/sting/gatk/walkers/FlagStat.java
index e881dcab7..b0cc3b12a 100644
--- a/public/java/src/org/broadinstitute/sting/gatk/walkers/FlagStat.java
+++ b/public/java/src/org/broadinstitute/sting/gatk/walkers/FlagStat.java
@@ -45,12 +45,12 @@ import java.text.NumberFormat;
  */
 @DocumentedGATKFeature( groupName = "Quality Control and Simple Analysis Tools", extraDocs = {CommandLineGATK.class} )
 @Requires({DataSource.READS})
-public class FlagStat extends ReadWalker<Integer, Integer> {
+public class FlagStat extends ReadWalker<FlagStat.FlagStatus, FlagStat.FlagStatus> implements TreeReducible<FlagStat.FlagStatus> {
     @Output
     PrintStream out;
 
     // what comes out of the flagstat
-    static class FlagStatus {
+    public final static class FlagStatus {
         long readCount = 0L;
         long QC_failure = 0L;
         long duplicates = 0L;
@@ -117,62 +117,89 @@ public class FlagStat extends ReadWalker<Integer, Integer> {
             return builder.toString();
         }
 
-    }
+        public FlagStatus add(final FlagStatus other) {
+            readCount += other.readCount;
+            QC_failure += other.QC_failure;
+            duplicates += other.duplicates;
+            mapped += other.mapped;
+            paired_in_sequencing += other.paired_in_sequencing;
+            read1 += other.read1;
+            read2 += other.read2;
+            properly_paired += other.properly_paired;
+            with_itself_and_mate_mapped += other.with_itself_and_mate_mapped;
+            singletons += other.singletons;
+            with_mate_mapped_to_a_different_chr += other.with_mate_mapped_to_a_different_chr;
+            with_mate_mapped_to_a_different_chr_maq_greaterequal_than_5 += other.with_mate_mapped_to_a_different_chr_maq_greaterequal_than_5;
 
-
-    private FlagStatus myStat = new FlagStatus();
-
-    public Integer map( ReferenceContext ref, GATKSAMRecord read, ReadMetaDataTracker metaDataTracker ) {
-        myStat.readCount++;
-        if (read.getReadFailsVendorQualityCheckFlag()) {
-            myStat.QC_failure++;
+            return this;
         }
-        if (read.getDuplicateReadFlag()) {
-            myStat.duplicates++;
-        }
-        if (!read.getReadUnmappedFlag()) {
-            myStat.mapped++;
-        }
-        if (read.getReadPairedFlag()) {
-            myStat.paired_in_sequencing++;
 
-            if (read.getSecondOfPairFlag()) {
-                myStat.read2++;
-            } else if (read.getReadPairedFlag()) {
-                myStat.read1++;
+        public FlagStatus add(final GATKSAMRecord read) {
+            this.readCount++;
+
+            if (read.getReadFailsVendorQualityCheckFlag()) {
+                this.QC_failure++;
             }
-            if (read.getProperPairFlag()) {
-                myStat.properly_paired++;
+            if (read.getDuplicateReadFlag()) {
+                this.duplicates++;
             }
-            if (!read.getReadUnmappedFlag() && !read.getMateUnmappedFlag()) {
-                myStat.with_itself_and_mate_mapped++;
+            if (!read.getReadUnmappedFlag()) {
+                this.mapped++;
+            }
+            if (read.getReadPairedFlag()) {
+                this.paired_in_sequencing++;
 
-                if (!read.getReferenceIndex().equals(read.getMateReferenceIndex())) {
-                    myStat.with_mate_mapped_to_a_different_chr++;
+                if (read.getSecondOfPairFlag()) {
+                    this.read2++;
+                } else if (read.getReadPairedFlag()) {
+                    this.read1++;
+                }
+                if (read.getProperPairFlag()) {
+                    this.properly_paired++;
+                }
+                if (!read.getReadUnmappedFlag() && !read.getMateUnmappedFlag()) {
+                    this.with_itself_and_mate_mapped++;
 
-                    if (read.getMappingQuality() >= 5) {
-                        myStat.with_mate_mapped_to_a_different_chr_maq_greaterequal_than_5++;
+                    if (!read.getReferenceIndex().equals(read.getMateReferenceIndex())) {
+                        this.with_mate_mapped_to_a_different_chr++;
+
+                        if (read.getMappingQuality() >= 5) {
+                            this.with_mate_mapped_to_a_different_chr_maq_greaterequal_than_5++;
+                        }
                     }
                 }
+                if (!read.getReadUnmappedFlag() && read.getMateUnmappedFlag()) {
+                    this.singletons++;
+                }
             }
-            if (!read.getReadUnmappedFlag() && read.getMateUnmappedFlag()) {
-                myStat.singletons++;
-            }
+
+            return this;
         }
-        return 1;
-
     }
 
-    public Integer reduceInit() {
-        return 0;
+
+    @Override
+    public FlagStatus map( final ReferenceContext ref, final GATKSAMRecord read, final ReadMetaDataTracker metaDataTracker ) {
+        return new FlagStatus().add(read);
+   }
+
+    @Override
+    public FlagStatus reduceInit() {
+        return new FlagStatus();
     }
 
-    public Integer reduce(Integer value, Integer sum) {
-        return value + sum;
+    @Override
+    public FlagStatus reduce(final FlagStatus value, final FlagStatus sum) {
+        return sum.add(value);
     }
 
-    public void onTraversalDone(Integer result) {
-        //out.println("[REDUCE RESULT] Traversal result is: " + result);
-        out.println(myStat.toString());
+    @Override
+    public FlagStatus treeReduce(final FlagStatus value, final FlagStatus sum) {
+        return sum.add(value);
+    }
+
+    @Override
+    public void onTraversalDone(final FlagStatus result) {
+        out.println(result.toString());
     }
 }
\ No newline at end of file