/* 
 * Copyright 2015 Alec Ten Harmsel
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/*
 * A BAM summary similar to `samtools flagstat`
 */

package com.alectenharmsel.examples.hadoop;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.seqdoop.hadoop_bam.BAMInputFormat;
import org.seqdoop.hadoop_bam.SAMRecordWritable;
import htsjdk.samtools.SAMRecord;

public class BAMSummary extends Configured implements Tool
{
    public static String getRecordName(String longName)
    {
        String[] name = longName.split("\\.");
        if (name.length == 0)
        {
            return "unnamed";
        }

        return name[0];
    }

    public static class Map extends Mapper<LongWritable, SAMRecordWritable, Text, LongWritable>
    {
        public void map(LongWritable key, SAMRecordWritable contents, Context context) throws IOException, InterruptedException
        {
            SAMRecord rec = contents.get();
            Boolean proper = false;
            String name = getRecordName(rec.getReadName());

            try
            {
                proper = rec.getProperPairFlag();
            }
            catch (IllegalStateException e)
            {
            }

            if (!proper)
            {
                context.write(new Text(name + "-improper"), new LongWritable(1));
            }

            context.write(new Text(name + "-total"), new LongWritable(1));
        }
    }

    public static class Reduce extends Reducer<Text, LongWritable, Text, LongWritable>
    {
        public void reduce(Text key, Iterable<LongWritable> vals, Context context) throws IOException, InterruptedException
        {
            long sum = 0;

            for(LongWritable tmp : vals)
            {
                sum += tmp.get();
            }

            context.write(key, new LongWritable(sum));
        }
    }

    public int run(String[] args) throws Exception
    {
        if(args.length != 2)
        {
            System.err.println("Usage: BAMSummary <input> <output>");
            System.exit(-1);
        }

        Job job = Job.getInstance(getConf(), "BAMSummary");
        job.setJarByClass(BAMSummary.class);

        job.setInputFormatClass(BAMInputFormat.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(Map.class);
        job.setCombinerClass(Reduce.class);
        job.setReducerClass(Reduce.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception
    {
        GenericOptionsParser parse = new GenericOptionsParser(new Configuration(), args);
        Configuration conf = parse.getConfiguration();

        int res = ToolRunner.run(conf, new BAMSummary(), parse.getRemainingArgs());

        System.exit(res);
    }
}

@@ -1753,16 +1753,24 @@
  extends
+%0A       
  Mapper%3C
@@ -1897,16 +1897,32 @@
 ontents,
+%0A               
  Context
@@ -1942,32 +1942,47 @@
 ows IOException,
+%0A              
  InterruptedExce
@@ -2132,24 +2132,76 @@
 ReadName());
+%0A            LongWritable one = new LongWritable(1);
 %0A%0A          
@@ -2269,16 +2269,436 @@
 rFlag();
+%0A%0A                if (proper)%0A                %7B%0A                        context.write(new Text(name + %22-properlypaired%22), one);%0A                %7D%0A%0A                if (rec.getSecondOfPairFlag())%0A                %7B%0A                    context.write(new Text(name + %22-read2%22), one);%0A                %7D%0A                else%0A                %7B%0A                    context.write(new Text(name + %22-read1%22), one);%0A                %7D
 %0A       
@@ -2797,15 +2797,34 @@
 if (
-!proper
+rec.getDuplicateReadFlag()
 )%0A  
@@ -2887,41 +2887,81 @@
 + %22-
-improper
+duplicates
 %22), 
+o
 ne
-w LongWritable(1));
+);%0A            %7D%0A%0A            if (rec.isValid() != null)
 %0A   
@@ -2961,35 +2961,38 @@
 ll)%0A            
-%7D%0A%0A
+%7B%0A    
             cont
@@ -3023,36 +3023,96 @@
 + %22-
-total
+invalid
 %22), 
+o
 ne
-w LongWritable(1)
+);%0A            %7D%0A%0A            context.write(new Text(name + %22-total%22), one
 );%0A 
@@ -3165,16 +3165,28 @@
  extends
+%0A           
  Reducer
@@ -3297,16 +3297,40 @@
 e%3E vals,
+%0A                       
  Context
@@ -3358,16 +3358,31 @@
 ception,
+%0A              
  Interru
@@ -4514,16 +4514,41 @@
 sParser(
+%0A                        
 new Conf
@@ -4679,16 +4679,40 @@
 mmary(),
+%0A                       
  parse.g

Add more statistics to BAMSummary