/*
 * Copyright (c) 2005-2012 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2014-11-12 23:17:08 +0000 (Wed, 12 Nov 2014) $' 
 * '$Revision: 33064 $'
 * 
 * Permission is hereby granted, without written agreement and without
 * license or royalty fees, to use, copy, modify, and distribute this
 * software and its documentation for any purpose, provided that the above
 * copyright notice and the following two paragraphs appear in all copies
 * of this software.
 *
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
 * ENHANCEMENTS, OR MODIFICATIONS.
 *
 */
package org.kepler.bio.stratosphere.input;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.kepler.ddp.Utilities;

import eu.stratosphere.api.common.io.statistics.BaseStatistics;
import eu.stratosphere.api.java.record.io.FileInputFormat;
import eu.stratosphere.configuration.Configuration;
import eu.stratosphere.core.fs.FileInputSplit;
import eu.stratosphere.core.fs.Path;
import eu.stratosphere.types.Record;
import eu.stratosphere.types.StringValue;



/**
* This Input Format is to read fasta sequence file and generate key value pair records.
* The format of each key is : partitionNumber_(readPosIndex+start) and each value is a list of sequences. 
* Parameter 'KEPLER::parameter::SequenceNumberPerExecution' specify at most how many sequences will be put into each record.
* This Input Format can guarantee each sequence is read in a whole and each record won't have truncated sequences.  
* It is built based on eu.stratosphere.pact.common.io.TextInputFormat class.
* 
* @author Jianwu Wang
* @version $Id: FASTAInputFormat.java 33064 2014-11-12 23:17:08Z crawl $
*/
public class FASTAInputFormat extends FileInputFormat
{
        /**
         * The configuration key to set the record delimiter.
         */
        public static final String RECORD_DELIMITER = "textformat.delimiter";
        
        /**
         * The configuration key to set the number of samples to take for the statistics.
         */
        public static final String NUM_STATISTICS_SAMPLES = "textformat.numSamples";

        /**
         * The configuration key to set the sequence number. It specifies the sequence number to be processed for each sub-workflow execution.
         */
        public static final String SEQUENCE_NUM = Utilities.CONFIGURATION_KEPLER_PARA_PREFIX + "::SequenceNumberPerExecution";
        
        /**
         * The configuration key to set position of the record delimiter. its value is null, prefix or suffix.
         */
        public static final String DELIMITER_POSITION = "textformat.delimiter.position";
        
        /**
         * The log.
         */
        private static final Log LOG = LogFactory.getLog(FASTAInputFormat.class);
        
        /**
         * The default read buffer size = 1MB.
         */
        private static final int DEFAULT_READ_BUFFER_SIZE = 1024 * 1024;
        
        /**
         * The default number of sample lines to consider when calculating the line width.
         */
        private static final int DEFAULT_NUM_SAMPLES = 10;
        
        // --------------------------------------------------------------------------------------------
        
        private byte[] readBuffer;

        private byte[] wrapBuffer;

        private int readPos;
        
        //the index of the current read position, it is used to be the key of record.
        private long readPosIndex;
        
        private int seqIndex;
        
        private long localSplitLength;
        
        private int partitionNumber;

        private int limit;

        //set delimiter to be '>' for fasta file
        private byte[] delimiter = new byte[] { '>' };

        private boolean overLimit;

        private boolean end;
        
        private int bufferSize = -1;
        
        private int numLineSamples;                                                                             // the number of lines to sample for statistics
        
        Integer seq_number = null;
        
        private long readPosOffset;
        
        String delimiter_position = null;
        
        private Path splitPath;


        // --------------------------------------------------------------------------------------------

        /**
         * Gets the delimiter that defines the record boundaries.
         * 
         * @return The delimiter, as bytes.
         */
        public byte[] getDelimiter()
        {
                return delimiter;
        }
        
        /**
         * Sets the size of the buffer to be used to find record boundaries. This method has only an effect, if it is called
         * before the input format is opened.
         * 
         * @param bufferSize The buffer size to use.
         */
        public void setBufferSize(int bufferSize)
        {
                this.bufferSize = bufferSize;
        }
        
        /**
         * Gets the size of the buffer internally used to parse record boundaries.
         * 
         * @return The size of the parsing buffer.
         */
        public int getBufferSize()
        {
                return this.readBuffer == null ? 0: this.readBuffer.length;
        }
        
        // --------------------------------------------------------------------------------------------
        
        /**
         * Configures this input format by reading the path to the file from the configuration and the string that
         * defines the record delimiter.
         * 
         * @param parameters The configuration object to read the parameters from.
         */
        @Override
        public void configure(Configuration parameters)
        {
                super.configure(parameters);
                //set delimiter to be '>' for fasta file
                String delimString = parameters.getString(RECORD_DELIMITER, ">");
                System.out.println("delimString:" + delimString);
                if (delimString == null) {
                        throw new IllegalArgumentException("The delimiter not be null.");
                }

                this.delimiter = delimString.getBytes();
                
                // set the number of samples
                this.numLineSamples = DEFAULT_NUM_SAMPLES;
                String samplesString = parameters.getString(NUM_STATISTICS_SAMPLES, null);
                
                if (samplesString != null) {
                        try {
                                this.numLineSamples = Integer.parseInt(samplesString);
                        }
                        catch (NumberFormatException nfex) {
                                if (LOG.isWarnEnabled())
                                        LOG.warn("Invalid value for number of samples to take: " + samplesString +
                                                        ". Using default value of " + DEFAULT_NUM_SAMPLES);
                        }
                }
                
                this.seq_number = new Integer(parameters.getString(SEQUENCE_NUM, "5000"));
                
                System.out.println("SEQUENCE_NUM:"+this.seq_number);
                
                delimiter_position = parameters.getString(DELIMITER_POSITION, "prefix");
                
        }
        
        
        
        // --------------------------------------------------------------------------------------------

        /* (non-Javadoc)
         * @see eu.stratosphere.pact.common.io.InputFormat#getStatistics()
         */
        @Override
        public FileBaseStatistics getStatistics(BaseStatistics cachedStatistics)
        {
                return null;
        }

        /**
         * Opens the given input split. This method opens the input stream to the specified file, allocates read buffers
         * and positions the stream at the correct position, making sure that any partial record at the beginning is skipped.
         * 
         * @param split The input split to open.
         * 
         * @see eu.stratosphere.pact.common.io.FileInputFormat#open(eu.stratosphere.nephele.fs.FileInputSplit)
         */
        @Override
        public void open(FileInputSplit split) throws IOException
        {
                super.open(split);
                this.splitPath = split.getPath();
                partitionNumber = split.getSplitNumber();
                localSplitLength = split.getLength();
                readPosOffset = 0;
                readPosIndex = 0;
                
                //System.out.println("split in one open():" + split);
                
                this.bufferSize = this.bufferSize <= 0 ? DEFAULT_READ_BUFFER_SIZE : this.bufferSize;
                this.readBuffer = new byte[this.bufferSize];
                this.wrapBuffer = new byte[256];

                this.limit = 0;
                this.readPos = 0;
                this.overLimit = false;
                this.end = false;

                if (this.splitStart != 0) {
                //the split is not the first split, so goes back to the last split and gets its last record.
                        this.stream.seek(this.splitStart);                      
                        readPosIndex = readSequence(1).length;
                        
                        // if the first partial record already pushes the stream over the limit of our split, then no
                        // record starts within this split 
                        if (this.overLimit) {
                                this.end = true;
                        }
                }
                else {
                        fillBuffer();
                }
        }

        /**
         * Checks whether the current split is at its end.
         * 
         * @return True, if the split is at its end, false otherwise.
         */
        @Override
        public boolean reachedEnd()
        {
                return this.end;
        }
        
//      @Override
//      public boolean nextRecord(KeyValuePair<PactString, PactString> pair) throws IOException
//      {
//              byte[] seqBytes = readSequence(seq_number.intValue());
//              if (seqBytes == null) {
//                      this.end = true;
//              }
//              if (end)
//                      return false;
//              pair.setKey(new PactString (new Integer(this.partitionNumber).toString() + "_" + new Long(this.readPosIndex + this.start).toString()));
//              pair.setValue(new PactString(new String(seqBytes)));
//              System.out.println("value size for key: " + pair.getKey() + " is " + seqBytes.length);
//              //update readPosIndex, which will be the key of the next record.
//              readPosIndex = readPosIndex + seqBytes.length;
//              return true;
//      }
        
        @Override
        public boolean nextRecord(Record record) throws IOException {
                final StringValue string = new StringValue();
                byte[] seqBytes = readSequence(seq_number.intValue());
                if (seqBytes == null) {
                        this.end = true;
                }
                if (end)
                        return false;
                record.setField(0, new StringValue (splitPath.getName() + "_" + new Integer(this.partitionNumber).toString() + "_" + new Long(this.readPosIndex + this.splitStart).toString()));
                string.setValueAscii(seqBytes, 0, seqBytes.length);
                record.setField(1, string);
                //System.out.println("value size for key: " + record.getField(0, PactString.class) + " is " + record.getField(1, PactString.class));
                //update readPosIndex, which will be the key of the next record.
                readPosIndex = readPosIndex + seqBytes.length;
                return true;
        }

        /**
         * Closes the input by releasing all buffers and closing the file input stream.
         * 
         * @throws IOException Thrown, if the closing of the file stream causes an I/O error.
         */
        @Override
        public void close() throws IOException
        {
                this.wrapBuffer = null;
                this.readBuffer = null;
                this.readPosIndex = 0;
                
                super.close();
        }

        // --------------------------------------------------------------------------------------------

        private byte[] readSequence(int seqNumber) throws IOException {
                if (this.stream == null || this.overLimit) {
                        return null;
                }
                //System.out.println("just enter into out readSequence. this.readPos: " + this.readPos + ", this.limit:" + this.limit + ", seqIndex:" + seqIndex + ", seqNumber:" + seqNumber + ", readPosOffset:" + readPosOffset + ", localSplitLength:" + localSplitLength);

                
                int countInWrapBuffer = 0;

                /* position of matching positions in the delimiter byte array */
                int i = 0;
                
                //it start with 1 because that the first delimiter is be counted, since the '>' is always at the beginning of data.
                seqIndex = 1;

                while (true) {
                        if (this.readPos >= this.limit) {
                                if (!fillBuffer()) {
                                        if (countInWrapBuffer > 0) {
                                                byte[] tmp = new byte[countInWrapBuffer];
                                                System.arraycopy(this.wrapBuffer, 0, tmp, 0, countInWrapBuffer);
                                                return tmp;
                                        } else {
                                                return null;
                                        }
                                }
                        }

                        int startPos = this.readPos;
                        int count = 0;

                        //System.out.println("before while. this.readPos: " + this.readPos + ", this.limit:" + this.limit + ", seqIndex:" + seqIndex + ", seqNumber:" + seqNumber + ", readPosOffset:" + readPosOffset + ", localSplitLength:" + localSplitLength);

                        while (this.readPos < this.limit && seqIndex <= seqNumber && readPosOffset < localSplitLength) {
                                
                                if ((this.readBuffer[this.readPos]) == this.delimiter[i] && readPos - startPos > this.delimiter.length) {
                                //the second condition is to make sure the first delimiter will not be counted, since the '>' is always at the beginning of data.
                                        i++;
                                        if (i == this.delimiter.length) {
                                                seqIndex++;
                                                i = 0;
                                        }
                                } else {
                                        i = 0;
                                }
                                //readPosOffset add one because readPos add one.
                                readPosOffset++;
                                readPos++;
                        }

                        //System.out.println("drop out reason. this.readPos: " + this.readPos + ", this.limit:" + this.limit + ", seqIndex:" + seqIndex + ", seqNumber:" + seqNumber + ", readPosOffset:" + readPosOffset + ", localSplitLength:" + localSplitLength);
                        // check why we dropped out
                        if (this.readPos == this.limit) {
                                count = this.limit - startPos;
                                // buffer exhausted
                                while (this.wrapBuffer.length - countInWrapBuffer < count) {
                                        // reallocate
                                        byte[] tmp = new byte[this.wrapBuffer.length * 2];
                                        System.arraycopy(this.wrapBuffer, 0, tmp, 0, countInWrapBuffer);
                                        this.wrapBuffer = tmp;
                                }

                                System.arraycopy(this.readBuffer, startPos, this.wrapBuffer, countInWrapBuffer, count);
                                countInWrapBuffer += count;
                        }
                        //if (seqIndex == seqNumber || readPosOffset > splitLength) {
                        else {
                                if (readPosOffset == localSplitLength) {
                                //reach the end of the split. Now read the across the boundary until the first delimiter of the next split.
                                        //System.out.println("reach end of the split");
                                        i = 0;
                                        while (this.readPos < this.limit && i < this.delimiter.length) {
                                                if ((this.readBuffer[this.readPos++]) == this.delimiter[i]) {
                                                        i++;
                                                } else {
                                                        i = 0;
                                                }

                                        }
                                        //set overLimit to be true since the end of this split has been reached.
                                        this.overLimit = true;
                                } else {
                                        //reach the next sequence. the index will be subtracted by 1 since its initial value is 1.
                                        seqIndex--;
                                }
                                //this.delimiter.length need to be subtracted from readPos because we do not want to count the beginning '>' of the next sequence.
                                count = this.readPos - startPos - this.delimiter.length;
                                //move read position back before the delimiter.
                                this.readPos = this.readPos - this.delimiter.length;
                                this.readPosOffset = this.readPosOffset - this.delimiter.length;

                                // copy to byte array
                                if (countInWrapBuffer > 0) {
                                        byte[] end = new byte[countInWrapBuffer + count];
                                        if (count >= 0) {
                                                System.arraycopy(this.wrapBuffer, 0, end, 0, countInWrapBuffer);
                                                System.arraycopy(this.readBuffer, 0, end, countInWrapBuffer, count);
                                                return end;
                                        } else {
                                                // count < 0
                                                System.arraycopy(this.wrapBuffer, 0, end, 0, countInWrapBuffer + count);
                                                return end;
                                        }
                                } else {
                                        byte[] end = new byte[count];
                                        System.arraycopy(this.readBuffer, startPos, end, 0, count);
                                        return end;
                                }
                        } 
                }
        }
        
        private final boolean fillBuffer() throws IOException {
                int toRead = this.splitLength > this.readBuffer.length ? this.readBuffer.length : (int) this.splitLength;
                if (this.splitLength <= 0) {
                        toRead = this.readBuffer.length;
                        this.overLimit = true;
                }

                int read = this.stream.read(this.readBuffer, 0, toRead);

                if (read == -1) {
                        this.stream.close();
                        this.stream = null;
                        return false;
                } else {
                        this.splitLength -= read;
                        this.readPos = 0;
                        this.limit = read;
                        return true;
                }
        }
        
}