package edu.ohsu.sonmezsysbio.cloudbreak.mapper;

import edu.ohsu.sonmezsysbio.cloudbreak.AlignmentRecord;
import edu.ohsu.sonmezsysbio.cloudbreak.io.SAMAlignmentReader;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.apache.log4j.Logger;

import java.io.*;
import java.util.zip.GZIPOutputStream;

/**
 * Created by IntelliJ IDEA.
 * User: cwhelan
 * Date: 6/12/13
 * Time: 4:17 PM
 */
public abstract class AlignerMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
    private static org.apache.log4j.Logger logger = Logger.getLogger(SingleEndAlignerMapper.class);
    protected String localDir;

    protected File s1File;
    protected File s2File;
    protected Writer s1FileWriter;
    protected Writer s2FileWriter;
    protected OutputCollector<Text, Text> output;
    protected Reporter reporter;

    protected boolean getCompressTempReadFile() {
        return true;
    }

    @Override
    public void configure(JobConf job) {
        super.configure(job);
        this.localDir = job.get("mapred.child.tmp");
        try {
            FileAndWriter faw = openFileWriter(localDir, 1);
            s1File = faw.file;
            s1FileWriter = faw.writer;
            faw = openFileWriter(localDir, 2);
            s2File = faw.file;
            s2FileWriter = faw.writer;
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    protected int getNumRunningMappers() throws IOException {
        int numRunningMappers = 0;
        Runtime runtime = Runtime.getRuntime();
        String cmds[] = {"ps", "-C", getCommandName(), "--no-headers"};
        Process proc = runtime.exec(cmds);
        InputStream inputstream = proc.getInputStream();
        InputStreamReader inputstreamreader = new InputStreamReader(inputstream);
        BufferedReader bufferedreader = new BufferedReader(inputstreamreader);
        while (bufferedreader.readLine() != null) {
            numRunningMappers++;
        }

        return numRunningMappers;
    }

    protected abstract String getCommandName();

    protected void waitForOpenSlot(int maxProcessesOnNode, Reporter reporter) throws IOException, InterruptedException {
        while (true) {
            // sleep for a random length of time between 0 and 60 seconds
            long sleepTime = (long) (Math.random() * 1000 * 60);
            logger.info("sleeping for " + sleepTime);
            Thread.sleep(sleepTime);
            int numRunningMappers = getNumRunningMappers();
            logger.info("num running mappers: " + numRunningMappers);
            if (numRunningMappers < maxProcessesOnNode) return;
            reporter.progress();
        }
    }

    protected void readAlignments(BufferedReader stdInput, InputStream errorStream) throws IOException {
        String outLine;
        SAMAlignmentReader alignmentReader = new SAMAlignmentReader();
        while ((outLine = stdInput.readLine()) != null) {
            if (logger.isDebugEnabled()) {
                logger.debug("LINE: " + outLine);
            }
            if (outLine.startsWith("@"))  {
                logger.debug("SAM HEADER LINE: " + outLine);
                continue;
            }

            String readPairId = outLine.substring(0,outLine.indexOf('\t')-2);
            AlignmentRecord alignment = alignmentReader.parseRecord(outLine);

            if (! alignment.isMapped()) {
                continue;
            }

            getOutput().collect(new Text(readPairId), new Text(outLine));

        }

        String errLine;
        BufferedReader errorReader = new BufferedReader(new InputStreamReader(errorStream));
        while ((errLine = errorReader.readLine()) != null) {
            logger.error("ERROR: " + errLine);
        }
    }

    protected String printErrorStream(InputStream errorStream) throws IOException {
        String outLine;BufferedReader stdErr = new BufferedReader(new
                InputStreamReader(errorStream));
        String firstErrorLine = null;
        while ((outLine = stdErr.readLine()) != null) {
            if (firstErrorLine == null) firstErrorLine = outLine;
            logger.error(outLine);
        }
        return firstErrorLine;
    }

    static class FileAndWriter {
        File file;
        Writer writer;

        FileAndWriter(File file, Writer writer) {
            this.file = file;
            this.writer = writer;
        }
    }

    protected FileAndWriter openFileWriter(String localDir, int fileNum) throws IOException {
        File outFile;
        Writer outWriter;
        if (getCompressTempReadFile()) {
            outFile = new File(localDir + "/temp" + fileNum + "_sequence.fastq.gz").getAbsoluteFile();
            outFile.createNewFile();
            outWriter = new OutputStreamWriter(new GZIPOutputStream(new FileOutputStream(outFile)));
        } else {
            outFile = new File(localDir + "/temp" + fileNum + "_sequence.fastq").getAbsoluteFile();
            outFile.createNewFile();
            outWriter = new OutputStreamWriter(new FileOutputStream(outFile));
        }

        return new FileAndWriter(outFile, outWriter);
    }

    public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
        if (this.output == null) {
            this.output = output;
        }
        if (this.reporter == null) {
            this.reporter = reporter;
        }

        String line = value.toString();
        String[] reads = line.split("\n");
        splitRead(key, reads[0], s1FileWriter);
        splitRead(key, reads[1], getRead2FileWriter());

        reporter.progress();
    }

    protected abstract Writer getRead2FileWriter();

    private void splitRead(LongWritable key, String line, Writer writer) throws IOException {
        String[] fields = line.split("\t");

        if (fields[1].length() != fields[3].length()) {
            logger.warn("Warning; mismatching seq and qual lengths in record " + key.toString() + "!");
            logger.warn("Seq:");
            logger.warn(fields[1]);
            logger.warn("Qual:");
            logger.warn(fields[3]);
            logger.warn("DONE WARNING");
        }
        writer.write(fields[0] + "\n");
        writer.write(fields[1] + "\n");
        writer.write(fields[2] + "\n");
        writer.write(fields[3] + "\n");
    }

    public OutputCollector<Text, Text> getOutput() {
        return output;
    }

    public void setOutput(OutputCollector<Text, Text> output) {
        this.output = output;
    }

    public Reporter getReporter() {
        return reporter;
    }

    public void setReporter(Reporter reporter) {
        this.reporter = reporter;
    }

    @Override
    public void close() throws IOException {
        super.close();
        if (s1FileWriter != null) s1FileWriter.close();
        if (s2FileWriter != null) s2FileWriter.close();
    }
}

@@ -270,24 +270,55 @@
 p.mapred.*;%0A
+import org.apache.log4j.Level;%0A
 import org.a
@@ -3367,12 +3367,177 @@
 %5Ct')
--2
 );
+%0A            if (readPairId.endsWith(%22/1%22) %7C%7C readPairId.endsWith(%22/2%22)) %7B%0A                readPairId = readPairId.substring(0, readPairId.length() - 2);%0A            %7D
 %0A   

fix issue with read names with whitespace