public class IntArrayWritable extends BinaryComparable implements WritableComparable<BinaryComparable> {

    private static final Log LOG = LogFactory.getLog(IntArrayWritable.class);
    
    private int[] intArray;
    
    private byte[] prevBytes;
    
    public IntArrayWritable() {}
    
    public IntArrayWritable(IntArrayWritable writable) { set(writable); }
    
    public IntArrayWritable(int[] intArray) { set(intArray); }
    
    public IntArrayWritable(List<Integer> intArray) { set(intArray); }
    
    /**
     * Set the value.
     */
    public void set(int[] intArray) {
        this.intArray = intArray;
        this.prevBytes = null;
    }
    
    public void set(List<Integer> intArray) {
        int[] arr = new int[intArray.size()];
        for(int i=0;i<intArray.size();i++) {
            arr[i] = intArray.get(i);    
        }
        this.intArray = arr;
        this.prevBytes = null;
    }
    
    public void set(IntArrayWritable that) {
        this.intArray = that.intArray;
        this.prevBytes = that.prevBytes;
    }
    
    public void setEmpty() {
        this.intArray = null;
        this.prevBytes = null;
    }
    
    public boolean isEmpty() {
        if(this.intArray == null) {
            return true;
        }
        return false;
    }

    /**
     * Return the value.
     */
    public int[] get() {
        return this.intArray;
    }
    
    @Override
    public void readFields(DataInput in) throws IOException {
        int count = in.readInt();
        
        int[] arr = new int[count];
        for (int i = 0; i < count; i++) {
            arr[i] = in.readInt();
        }
        
        this.intArray = arr;
        this.prevBytes = null;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        int count = this.intArray.length;
        out.writeInt(count);
        
        for (int i = 0; i < count; i++) {
            out.writeInt(this.intArray[i]);
        }
    }
    
    /**
     * Returns true iff
     */
    @Override
    public boolean equals(Object o) {
        if (o instanceof IntArrayWritable) {
            return super.equals(o);
        }
        return false;
    }

    @Override
    public int hashCode() {
        return super.hashCode();
    }
    
    @Override
    public String toString() {
        String value = new String();
        
        for(int i=0;i<this.intArray.length;i++) {
            if(value.length() != 0) {
                value += ",";
            }
            value += this.intArray[i];
        }
        return value;
    }
    
    @Override
    public int getLength() {
        return this.intArray.length * 4;
    }

    @Override
    public byte[] getBytes() {
        if(this.prevBytes == null) {
            byte[] arr = new byte[this.intArray.length * 4];
            for(int i=0;i<this.intArray.length;i++) {
                int ivalue = this.intArray[i];
                arr[4*i] = (byte) ((ivalue >> 24) & 0xff);
                arr[4*i+1] = (byte) ((ivalue >> 16) & 0xff);
                arr[4*i+2] = (byte) ((ivalue >> 8) & 0xff);
                arr[4*i+3] = (byte) (ivalue & 0xff);
            }
            this.prevBytes = arr;
        }
        return prevBytes;
    }
    
    /** A Comparator optimized for IntArrayWritable. */ 
    public static class Comparator extends WritableComparator {

        public Comparator() {
            super(IntArrayWritable.class);
        }

        /**
         * Compare the buffers in serialized form.
         */
        @Override
        public int compare(byte[] b1, int s1, int l1,
                byte[] b2, int s2, int l2) {
            return compareBytes(b1, s1, l1,
                    b2, s2, l2);
        }
    }

    static {
        // register this comparator
        WritableComparator.define(IntArrayWritable.class, new IntArrayWritable.Comparator());
    }
} public class CompressedSequenceWritable extends BinaryComparable implements WritableComparable<BinaryComparable> {

    private static final Log LOG = LogFactory.getLog(CompressedSequenceWritable.class);
    
    private byte[] compressedSequence;
    private int seqLength;
    
    private static final int LENGTH_BYTES = 1;
    
    public CompressedSequenceWritable() {}
    
    public CompressedSequenceWritable(String sequence) throws IOException { set(sequence); }
    
    public CompressedSequenceWritable(byte[] compressedSequence, int seqLength) { set(compressedSequence, seqLength); }
    
    /**
     * Set the value.
     */
    public void set(byte[] compressedSequence, int seqLength) {
        this.compressedSequence = compressedSequence;
        this.seqLength = seqLength;
    }
    
    public void set(String sequence) throws IOException {
        this.compressedSequence = SequenceHelper.compress(sequence);
        this.seqLength = sequence.length();
    }
    
    public void set(CompressedSequenceWritable that) throws IOException {
        this.compressedSequence = that.compressedSequence;
        this.seqLength = that.seqLength;
    }
    
    public void setEmpty() {
        this.compressedSequence = null;
        this.seqLength = 0;
    }

    public boolean isEmpty() {
        if(this.compressedSequence == null || this.seqLength == 0) {
            return true;
        }
        return false;
    }
    
    /**
     * Return the value.
     */
    public byte[] getCompressedSequence() {
        return this.compressedSequence;
    }
    
    public String getSequence() {
        return SequenceHelper.decompress(this.compressedSequence, this.seqLength);
    }
    
    public int getSequenceLength() {
        return this.seqLength;
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.seqLength = in.readByte();
        int byteLen = SequenceHelper.getCompressedSize(this.seqLength);
        this.compressedSequence = new byte[byteLen];
        in.readFully(this.compressedSequence, 0, byteLen);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeByte(this.seqLength);
        out.write(this.compressedSequence);
    }
    
    /**
     * Returns true iff
     */
    @Override
    public boolean equals(Object o) {
        if (o instanceof CompressedSequenceWritable) {
            return super.equals(o);
        }
        return false;
    }

    @Override
    public int hashCode() {
        return SequenceHelper.hashCode(this.compressedSequence);
    }

    @Override
    public String toString() {
        return SequenceHelper.decompress(this.compressedSequence, this.seqLength);
    }

    @Override
    public int getLength() {
        return this.compressedSequence.length;
    }

    @Override
    public byte[] getBytes() {
        return this.compressedSequence;
    }
    
    /** A Comparator optimized for CompressedSequenceWritable. */ 
    public static class Comparator extends WritableComparator {

        public Comparator() {
            super(CompressedSequenceWritable.class);
        }

        /**
         * Compare the buffers in serialized form.
         */
        public int compare(byte[] b1, int s1, int l1,
                byte[] b2, int s2, int l2) {
            return compareBytes(b1, s1 + LENGTH_BYTES, l1 - LENGTH_BYTES,
                    b2, s2 + LENGTH_BYTES, l2 - LENGTH_BYTES);
        }
    }

    static {
        // register this comparator
        WritableComparator.define(CompressedSequenceWritable.class, new Comparator());
    }
} public class FileSystemHelper {
    public static String makeCommaSeparated(Path[] strs) {
        if(strs == null) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        for(int i=0;i<strs.length;i++) {
            sb.append(strs[i].toString());
            if(i < strs.length - 1) {
                sb.append(",");
            }
        }
        return sb.toString();
    }
    
    public static String makeCommaSeparated(Collection<String> strs) {
        if(strs == null) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        int i=0;
        for(String str : strs) {
            sb.append(str);
            i++;
            
            if(i<strs.size()) {
                sb.append(",");
            }
        }
        return sb.toString();
    }
    
    public static String makeCommaSeparated(String[] strs) {
        if(strs == null) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        for(int i=0;i<strs.length;i++) {
            sb.append(strs[i]);
            if(i < strs.length - 1) {
                sb.append(",");
            }
        }
        return sb.toString();
    }
    
    public static String[] splitCommaSeparated(String comma_separated_input) {
        String[] inputs = comma_separated_input.split(",");
        return inputs;
    }
    
    public static Path[] makePathFromString(Configuration conf, Collection<String> pathStrings) throws IOException {
        return makePathFromString(conf, pathStrings.toArray(new String[0]));
    }
    
    public static Path[] makePathFromString(Configuration conf, String[] pathStrings) throws IOException {
        List<Path> paths = new ArrayList<Path>();
        for(String path : pathStrings) {
            if(hasWildcard(path)) {
                Path[] patharr = resolveWildcard(conf, path);
                for(Path pathentry : patharr) {
                    paths.add(pathentry);
                }
            } else {
                paths.add(new Path(path));
            }
        }
        return paths.toArray(new Path[0]);
    }
    
    public static String[] makeStringFromPath(Path[] paths) {
        String[] pathStrings = new String[paths.length];
        for(int i=0;i<paths.length;i++) {
            pathStrings[i] = paths[i].toString();
        }
        return pathStrings;
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, String inputPathsCommaSeparated) throws IOException {
        return FileSystemHelper.getAllSamplePaths(conf, makePathFromString(conf, splitCommaSeparated(inputPathsCommaSeparated)));
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, String[] inputPaths) throws IOException {
        return FileSystemHelper.getAllSamplePaths(conf, makePathFromString(conf, inputPaths));
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, Collection<String> inputPaths) throws IOException {
        return FileSystemHelper.getAllSamplePaths(conf, makePathFromString(conf, inputPaths));
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, Path[] inputPaths) throws IOException {
        List<Path> inputFiles = new ArrayList<Path>();
        SamplePathFilter filter = new SamplePathFilter();
        
        for(Path path : inputPaths) {
            FileSystem fs = path.getFileSystem(conf);
            FileStatus status = fs.getFileStatus(path);
            if(status.isDirectory()) {
                FileStatus[] entries = fs.listStatus(path);
                entries = sortFileStatusList(entries);
                for(FileStatus entry : entries) {
                    if(filter.accept(entry.getPath())) {
                        inputFiles.add(entry.getPath());
                    }
                }
            } else {
                FileStatus entry = fs.getFileStatus(path);
                if(filter.accept(entry.getPath())) {
                    inputFiles.add(entry.getPath());
                }
            }
        }
        
        Path[] files = inputFiles.toArray(new Path[0]);
        return files;
    }
    
    public static boolean hasWildcard(String path) {
        if(path.indexOf("*") >= 0) {
            return true;
        }
        
        return false;
    }

    private static Path[] resolveWildcard(Configuration conf, String path) throws IOException {
        List<Path> paths = new ArrayList<Path>();
        
        int idxWildcard = path.lastIndexOf("*");
        if(idxWildcard >= 0) {
            String left = null;
            String right = path.substring(idxWildcard+1);
            
            String parentPath = path.substring(0, idxWildcard);
            int idxParent = parentPath.lastIndexOf("/");
            if(idxParent >= 0) {
                parentPath = parentPath.substring(0, idxParent);
                left = path.substring(idxParent+1, idxWildcard);
            } else {
                parentPath = "";
                left = path.substring(0, idxWildcard);
            }
            
            Path parent = new Path(parentPath);
            FileSystem fs = parent.getFileSystem(conf);
            FileStatus status = fs.getFileStatus(parent);
            if(status.isDirectory()) {
                FileStatus[] entries = fs.listStatus(parent);
                entries = sortFileStatusList(entries);
                for(FileStatus entry : entries) {
                    if(!left.isEmpty()) {
                        if(!entry.getPath().getName().startsWith(left)) {
                            // skip
                            continue;
                        }
                    }
                    
                    if(!right.isEmpty()) {
                        if(!entry.getPath().getName().endsWith(right)) {
                            // skip
                            continue;
                        }
                    }
                    
                    paths.add(entry.getPath());
                }
            }
        }
        
        return paths.toArray(new Path[0]);
    }
    
    private static FileStatus[] sortFileStatusList(FileStatus[] list) {
        List<FileStatus> l = new ArrayList<FileStatus>();
        for(FileStatus status : list) {
            l.add(status);
        }
        
        Collections.sort(l, new Comparator<FileStatus>(){

            @Override
            public int compare(FileStatus arg0, FileStatus arg1) {
                return arg0.getPath().getName().compareTo(arg1.getPath().getName());
            }
        });
        
        return l.toArray(new FileStatus[0]);
    }
} public class Report {
    private List<Job> jobs;
    
    public Report() {
        this.jobs = new ArrayList<Job>();
    }

    public void addJob(org.apache.hadoop.mapreduce.Job job) {
        this.jobs.add(new Job(job));
    }
    
    public void addJob(Job job) {
        this.jobs.add(job);
    }
    
    public void addJob(Job[] jobs) {
        for(Job job : jobs) {
            this.jobs.add(job);
        }
    }
    
    public void addJob(List<Job> jobs) {
        this.jobs.addAll(jobs);
    }
    
    public void writeTo(String filename) throws IOException {
        writeTo(new File("./", filename));
    }
    
    public void writeTo(File f) throws IOException {
        if(f.getParentFile() != null) {
            if(!f.getParentFile().exists()) {
                f.getParentFile().mkdirs();
            }
        }
        
        Writer writer = new FileWriter(f, true);
        boolean first = true;
        for(Job job : this.jobs) {
            if(first) {
                first = false;
            }
            writer.write(makeText(job));
            writer.write("\n\n");
        }
        
        writer.close();
    }
    
    private String makeText(Job job) {
        return "Job : " + job.getName() + "\n" +
                "JobID : " + job.getID() + "\n" + 
                "Status : " + job.getStatus() + "\n" +
                "StartTime : " + TimeHelper.getTimeString(job.getStartTime()) + "\n" +
                "FinishTime : " + TimeHelper.getTimeString(job.getFinishTime()) + "\n" + 
                "TimeTaken : " + TimeHelper.getDiffTimeString(job.getStartTime(), job.getFinishTime()) + "\n\n" +
                job.getCountersStr();
    }
} public class SequenceFileInputFormat extends FileInputFormat<LongWritable, ReadInfo> {

    private static final Log LOG = LogFactory.getLog(SequenceFileInputFormat.class);
    
    @Override
    public RecordReader<LongWritable, ReadInfo> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        return new ReadRecordReader();
    }
    
    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        final CompressionCodec codec = new CompressionCodecFactory(context.getConfiguration()).getCodec(filename);
        if(codec == null) {
            return true;
        }
        
        return codec instanceof SplittableCompressionCodec;
    }
} public class PreprocessorConfigException extends Exception {
    static final long serialVersionUID = 7818375828146090155L;

    public PreprocessorConfigException() {
        super();
    }

    public PreprocessorConfigException(String string) {
        super(string);
    }

    public PreprocessorConfigException(String string, Throwable thrwbl) {
        super(string, thrwbl);
    }

    public PreprocessorConfigException(Throwable thrwbl) {
        super(thrwbl);
    }
} public class PreprocessorRoundConfig extends PreprocessorConfig {
    private FileTable fileTable;
    
    public static PreprocessorRoundConfig createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (PreprocessorRoundConfig) serializer.fromJsonFile(file, PreprocessorRoundConfig.class);
    }
    
    public static PreprocessorRoundConfig createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (PreprocessorRoundConfig) serializer.fromJson(json, PreprocessorRoundConfig.class);
    }
    
    public static PreprocessorRoundConfig createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (PreprocessorRoundConfig) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, PreprocessorRoundConfig.class);
    }
    
    public static PreprocessorRoundConfig createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (PreprocessorRoundConfig) serializer.fromJsonFile(fs, file, PreprocessorRoundConfig.class);
    }
    
    public PreprocessorRoundConfig() {
        
    }
    
    public PreprocessorRoundConfig(PreprocessorConfig config) {
        super(config);
    }
    
    @JsonProperty("file_table")
    public FileTable getFileTable() {
        return this.fileTable;
    }
    
    @JsonProperty("file_table")
    public void setFileTable(FileTable fileTable) {
        this.fileTable = fileTable;
    }
} public class FileTable {
    
    private static final Log LOG = LogFactory.getLog(FileTable.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.preprocess.common.filetable.filetable";
    
    private String name;
    private int kmerSize;
    private List<String> samples = new ArrayList<String>();
    private Hashtable<String, Integer> sampleIDCacheTable = new Hashtable<String, Integer>();
    
    public static FileTable createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJsonFile(file, FileTable.class);
    }
    
    public static FileTable createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJson(json, FileTable.class);
    }
    
    public static FileTable createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, FileTable.class);
    }
    
    public static FileTable createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJsonFile(fs, file, FileTable.class);
    }
    
    public FileTable() {
    }
    
    public FileTable(String name, int kmerSize) {
        this.name = name;
        this.kmerSize = kmerSize;
    }
    
    public FileTable(SampleGroup group, int kmerSize) {
        this.name = group.getName();
        this.kmerSize = kmerSize;
        
        Collection<SampleInfo> samples = group.getSamples();
        for(SampleInfo sample : samples) {
            this.samples.add(sample.getPath());
        }
    }
    
    @JsonProperty("name")
    public String getName() {
        return this.name;
    }
    
    @JsonProperty("name")
    public void setName(String name) {
        this.name = name;
    }
    
    @JsonProperty("kmer_size")
    public int getKmerSize() {
        return this.kmerSize;
    }
    
    @JsonProperty("kmer_size")
    public void setKmerSize(int kmerSize) {
        this.kmerSize = kmerSize;
    }
    
    @JsonIgnore
    public int samples() {
        return this.samples.size();
    }
    
    @JsonProperty("samples")
    public Collection<String> getSamples() {
        return this.samples;
    }
    
    @JsonIgnore
    public int getSampleID(String sample) {
        Integer sampleID = this.sampleIDCacheTable.get(sample);
        if(sampleID == null) {
            int sid = -1; // not found
            for(int i=0;i<this.samples.size();i++) {
                if(this.samples.get(i).compareTo(sample) == 0) {
                    sid = i;
                    break;
                }
            }
            
            if(sid < 0) {
                String sample_target_name = sample;
                int idx = sample_target_name.lastIndexOf("/");
                if(idx >= 0) {
                    sample_target_name = sample_target_name.substring(idx + 1);
                }
                
                for(int j=0;j<this.samples.size();j++) {
                    String sample_name = this.samples.get(j);
                    idx = sample_name.lastIndexOf("/");
                    if(idx >= 0) {
                        sample_name = sample_name.substring(idx + 1);
                    }
                    
                    if(sample_name.compareTo(sample_target_name) == 0) {
                        sid = j;
                        break;
                    }
                }
            }
            
            sampleID = sid;
            // cache
            this.sampleIDCacheTable.put(sample, sampleID);
        }
        return sampleID;
    }
    
    @JsonProperty("samples")
    public void addSample(Collection<String> samples) {
        this.samples.addAll(samples);
    }
    
    @JsonIgnore
    public void addSample(String sample) {
        this.samples.add(sample);
    }
    
    @JsonIgnore
    public void clearSample() {
        this.samples.clear();
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class KmerFilterHelper {
    
    private final static String KMER_FILTER_TABLE_PATH_EXP = ".+\\." + PreprocessorConstants.KMER_FILTER_TABLE_FILENAME_EXTENSION + "$";
    private final static Pattern KMER_FILTER_TABLE_PATH_PATTERN = Pattern.compile(KMER_FILTER_TABLE_PATH_EXP);
    
    private final static String KMER_FILTER_PART_TABLE_PATH_EXP = ".+\\." + PreprocessorConstants.KMER_FILTER_TABLE_FILENAME_EXTENSION + "\\.\\d+$";
    private final static Pattern KMER_FILTER_PART_TABLE_PATH_PATTERN = Pattern.compile(KMER_FILTER_PART_TABLE_PATH_EXP);
    
    public static String makeKmerFilterTableFileName(String filename) {
        return filename + "." + PreprocessorConstants.KMER_FILTER_TABLE_FILENAME_EXTENSION;
    }
    
    public static String makeKmerFilterPartTableFileName(String filename, int taskID) {
        return filename + "." + PreprocessorConstants.KMER_FILTER_TABLE_FILENAME_EXTENSION + "." + taskID;
    }
    
    public static String makeKmerFilterDirPath(String rootPath) {
        return PathHelper.concatPath(rootPath, PreprocessorConstants.KMER_FILTER_DIRNAME);
    }
    
    public static boolean isKmerFilterTableFile(Path path) {
        return isKmerFilterTableFile(path.getName());
    }
    
    public static boolean isKmerFilterTableFile(String path) {
        Matcher matcher = KMER_FILTER_TABLE_PATH_PATTERN.matcher(path.toLowerCase());
        if(matcher.matches()) {
            return true;
        }
        return false;
    }
    
    public static boolean isKmerFilterPartTableFile(Path path) {
        return isKmerFilterPartTableFile(path.getName());
    }
    
    public static boolean isKmerFilterPartTableFile(String path) {
        Matcher matcher = KMER_FILTER_PART_TABLE_PATH_PATTERN.matcher(path.toLowerCase());
        if(matcher.matches()) {
            return true;
        }
        return false;
    }
    
    public static Path[] getKmerFilterPartTableFilePaths(Configuration conf, Path inputPath) throws IOException {
        List<Path> inputFiles = new ArrayList<Path>();
        KmerFilterPartTablePathFilter filter = new KmerFilterPartTablePathFilter();
        
        FileSystem fs = inputPath.getFileSystem(conf);
        if(fs.exists(inputPath)) {
            FileStatus status = fs.getFileStatus(inputPath);
            if(status.isDirectory()) {
                // check child
                FileStatus[] entries = fs.listStatus(inputPath);
                for (FileStatus entry : entries) {
                    if(entry.isFile()) {
                        if (filter.accept(entry.getPath())) {
                            inputFiles.add(entry.getPath());
                        }
                    }
                }
            } else {
                if (filter.accept(inputPath)) {
                    inputFiles.add(inputPath);
                }
            }
        }
        
        Path[] files = inputFiles.toArray(new Path[0]);
        return files;
    }
} public class KmerFilter {
    
    private static final Log LOG = LogFactory.getLog(KmerFilter.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.preprocess.common.kmerfilter.kmerfilter";
    
    private String name;
    private long totalKmers;
    private long uniqueKmers;
    private long sumOfSquare;
    
    public static KmerFilter createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilter) serializer.fromJsonFile(file, KmerFilter.class);
    }
    
    public static KmerFilter createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilter) serializer.fromJson(json, KmerFilter.class);
    }
    
    public static KmerFilter createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilter) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerFilter.class);
    }
    
    public static KmerFilter createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilter) serializer.fromJsonFile(fs, file, KmerFilter.class);
    }
    
    public KmerFilter() {
    }
    
    public KmerFilter(String name) {
        this.name = name;
    }
    
    @JsonProperty("name")
    public String getName() {
        return this.name;
    }
    
    @JsonProperty("name")
    public void setName(String name) {
        this.name = name;
    }
    
    @JsonProperty("total_kmers")
    public void setTotalKmers(long totalKmers) {
        this.totalKmers = totalKmers;
    }
    
    @JsonProperty("total_kmers")
    public long getTotalKmers() {
        return this.totalKmers;
    }
    
    @JsonIgnore
    public void incrementTotalKmers(long totalKmers) {
        this.totalKmers += totalKmers;
    }
    
    @JsonProperty("unique_kmers")
    public void setUniqueKmers(long uniqueKmers) {
        this.uniqueKmers = uniqueKmers;
    }
    
    @JsonProperty("unique_kmers")
    public long getUniqueKmers() {
        return this.uniqueKmers;
    }
    
    @JsonIgnore
    public void incrementUniqueKmers(long uniqueKmers) {
        this.uniqueKmers += uniqueKmers;
    }
    
    @JsonProperty("sum_of_square")
    public void setSumOfSquare(long sumOfSquare) {
        this.sumOfSquare = sumOfSquare;
    }
    
    @JsonProperty("sum_of_square")
    public long getSumOfSquare() {
        return this.sumOfSquare;
    }
    
    @JsonIgnore
    public void incrementSumOfSquare(double sumOfSquare) {
        this.sumOfSquare += sumOfSquare;
    }
    
    @JsonIgnore
    public double getMean() {
        return (double)this.totalKmers / (double)this.uniqueKmers;
    }
    
    @JsonIgnore
    public double getStddev() {
        double mean = (double)this.totalKmers / (double)this.uniqueKmers;
        double mean_square = Math.pow(mean, 2);
        double term = (double)this.sumOfSquare / (double)this.uniqueKmers;
        return Math.sqrt(term - mean_square);
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class KmerFilterPart {
    
    private static final Log LOG = LogFactory.getLog(KmerFilterPart.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.preprocess.common.kmerfilter.kmerfilterpart";
    
    private String name;
    private long totalKmers;
    private long uniqueKmers;
    private long sumOfSquare;
    
    public static KmerFilterPart createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPart) serializer.fromJsonFile(file, KmerFilterPart.class);
    }
    
    public static KmerFilterPart createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPart) serializer.fromJson(json, KmerFilterPart.class);
    }
    
    public static KmerFilterPart createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPart) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerFilterPart.class);
    }
    
    public static KmerFilterPart createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPart) serializer.fromJsonFile(fs, file, KmerFilterPart.class);
    }
    
    public KmerFilterPart() {
    }
    
    public KmerFilterPart(String name) {
        this.name = name;
    }
    
    @JsonProperty("name")
    public String getName() {
        return this.name;
    }
    
    @JsonProperty("name")
    public void setName(String name) {
        this.name = name;
    }
    
    @JsonProperty("total_kmers")
    public void setTotalKmers(long totalKmers) {
        this.totalKmers = totalKmers;
    }
    
    @JsonProperty("total_kmers")
    public long getTotalKmers() {
        return this.totalKmers;
    }
    
    @JsonIgnore
    public void incrementTotalKmers(long totalKmers) {
        this.totalKmers += totalKmers;
    }
    
    @JsonProperty("unique_kmers")
    public void setUniqueKmers(long uniqueKmers) {
        this.uniqueKmers = uniqueKmers;
    }
    
    @JsonProperty("unique_kmers")
    public long getUniqueKmers() {
        return this.uniqueKmers;
    }
    
    @JsonIgnore
    public void incrementUniqueKmers(long uniqueKmers) {
        this.uniqueKmers += uniqueKmers;
    }
    
    @JsonProperty("sum_of_square")
    public void setSumOfSquare(long sumOfSquare) {
        this.sumOfSquare = sumOfSquare;
    }
    
    @JsonProperty("sum_of_square")
    public long getSumOfSquare() {
        return this.sumOfSquare;
    }
    
    @JsonIgnore
    public void incrementSumOfSquare(double sumOfSquare) {
        this.sumOfSquare += sumOfSquare;
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class KmerFilterPartTable {
    
    private static final Log LOG = LogFactory.getLog(KmerFilterPartTable.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.preprocess.common.kmerfilter.kmerfilterparttable";
    
    private String name;
    private List<KmerFilterPart> filterPart = new ArrayList<KmerFilterPart>();
    
    public static KmerFilterPartTable createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPartTable) serializer.fromJsonFile(file, KmerFilterPartTable.class);
    }
    
    public static KmerFilterPartTable createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPartTable) serializer.fromJson(json, KmerFilterPartTable.class);
    }
    
    public static KmerFilterPartTable createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPartTable) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerFilterPartTable.class);
    }
    
    public static KmerFilterPartTable createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterPartTable) serializer.fromJsonFile(fs, file, KmerFilterPartTable.class);
    }
    
    public KmerFilterPartTable() {
    }
    
    public KmerFilterPartTable(String name) {
        this.name = name;
    }
    
    @JsonProperty("name")
    public String getName() {
        return this.name;
    }
    
    @JsonProperty("name")
    public void setName(String name) {
        this.name = name;
    }
    
    @JsonIgnore
    public int filterPartNum() {
        return this.filterPart.size();
    }
    
    @JsonProperty("filter_part")
    public Collection<KmerFilterPart> getFilterPart() {
        return this.filterPart;
    }
    
    @JsonProperty("filter_part")
    public void addFilterPart(Collection<KmerFilterPart> filterPart) {
        this.filterPart.addAll(filterPart);
    }
    
    @JsonIgnore
    public void addFilterPart(KmerFilterPart filterPart) {
        this.filterPart.add(filterPart);
    }
    
    @JsonIgnore
    public void clearFilterPart() {
        this.filterPart.clear();
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class KmerFilterTable {
    
    private static final Log LOG = LogFactory.getLog(KmerFilterTable.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.preprocess.common.kmerfilter.kmerfiltertable";
    
    private String name;
    private List<KmerFilter> filter = new ArrayList<KmerFilter>();
    
    public static KmerFilterTable createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterTable) serializer.fromJsonFile(file, KmerFilterTable.class);
    }
    
    public static KmerFilterTable createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterTable) serializer.fromJson(json, KmerFilterTable.class);
    }
    
    public static KmerFilterTable createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterTable) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerFilterTable.class);
    }
    
    public static KmerFilterTable createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerFilterTable) serializer.fromJsonFile(fs, file, KmerFilterTable.class);
    }
    
    public KmerFilterTable() {
    }
    
    public KmerFilterTable(String name) {
        this.name = name;
    }
    
    @JsonProperty("name")
    public String getName() {
        return this.name;
    }
    
    @JsonProperty("name")
    public void setName(String name) {
        this.name = name;
    }
    
    @JsonIgnore
    public int filterNum() {
        return this.filter.size();
    }
    
    @JsonProperty("filter")
    public Collection<KmerFilter> getFilter() {
        return this.filter;
    }
    
    @JsonProperty("filter")
    public void addFilter(Collection<KmerFilter> filter) {
        this.filter.addAll(filter);
    }
    
    @JsonIgnore
    public void addFilter(KmerFilter filter) {
        this.filter.add(filter);
    }
    
    @JsonIgnore
    public void clearFilter() {
        this.filter.clear();
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
}
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
import libra.common.hadoop.io.datatypes.IntArrayWritable;
import libra.common.hadoop.io.datatypes.CompressedSequenceWritable;
import libra.common.helpers.FileSystemHelper;
import libra.common.report.Report;
import libra.common.hadoop.io.format.sequence.SequenceFileInputFormat;
import libra.preprocess.common.PreprocessorConfigException;
import libra.preprocess.common.PreprocessorRoundConfig;
import libra.preprocess.common.filetable.FileTable;
import libra.preprocess.common.helpers.KmerFilterHelper;
import libra.preprocess.common.kmerfilter.KmerFilter;
import libra.preprocess.common.kmerfilter.KmerFilterPart;
import libra.preprocess.common.kmerfilter.KmerFilterPartTable;
import libra.preprocess.common.kmerfilter.KmerFilterTable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
/*
 * Copyright 2016 iychoi.
 *
 * 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.
 */
package libra.preprocess.stage1;


/**
 *
 * @author iychoi
 */
public class KmerFilterBuilder {
    
    private static final Log LOG = LogFactory.getLog(KmerFilterBuilder.class);
    
    public KmerFilterBuilder() {
        
    }
    
    private void validatePreprocessorConfig(PreprocessorRoundConfig ppConfig) throws PreprocessorConfigException {