package org.opencb.hpg.bigdata.core.avro;

import org.opencb.biodata.models.variant.Variant;
import org.opencb.biodata.models.variant.VariantSource;
import org.opencb.biodata.models.variant.avro.VariantAvro;
import org.opencb.biodata.models.variant.avro.VariantFileMetadata;
import org.opencb.biodata.tools.variant.VariantVcfHtsjdkReader;
import org.opencb.biodata.tools.variant.stats.VariantGlobalStatsCalculator;
import org.opencb.hpg.bigdata.core.io.avro.AvroFileWriter;

import java.io.*;
import java.util.List;

/**
 * Created by jtarraga on 03/08/16.
 */
public class VariantAvroSerializer extends AvroSerializer<VariantAvro> {

    public VariantAvroSerializer() {
        super("deflate");
    }

    public VariantAvroSerializer(String compression) {
        super(compression);
    }

    public void toAvro(InputStream inputStream, String outputFilename) throws IOException {

        // reader
        String metaFilename = outputFilename + ".meta";
        VariantSource variantSource = new VariantSource(metaFilename, "0", "0", "s");
        VariantVcfHtsjdkReader vcfReader = new VariantVcfHtsjdkReader(inputStream, variantSource, null);
        vcfReader.open();
        vcfReader.pre();

        // writer
        OutputStream outputStream = new FileOutputStream(outputFilename);
        AvroFileWriter<VariantAvro> avroFileWriter = new AvroFileWriter<>(VariantAvro.SCHEMA$,
                                                                          compression, outputStream);
        avroFileWriter.open();
        VariantGlobalStatsCalculator statsCalculator = new VariantGlobalStatsCalculator(vcfReader.getSource());
        statsCalculator.pre();

        // main loop
        List<Variant> variants;
        while (true) {
            variants = vcfReader.read(1000);
            if (variants.size() == 0) {
                break;
            }
            // write variants and update stats
            for (Variant variant: variants) {
                if (filter(variant.getImpl())) {
                    avroFileWriter.writeDatum(variant.getImpl());
                    statsCalculator.updateGlobalStats(variant);
                }
            }
        }

        // write variant metadata (header, stats,...) in a separated file
        FileOutputStream out = new FileOutputStream(metaFilename);
        AvroFileWriter<VariantFileMetadata> aw = new AvroFileWriter<>(VariantFileMetadata.SCHEMA$,
                                                                      compression, out);
        aw.open();
        aw.writeDatum(variantSource.getImpl());
        aw.close();
        out.close();

        // close
        vcfReader.post();
        vcfReader.close();
        avroFileWriter.close();
        outputStream.close();
    }
}

@@ -36,16 +36,226 @@
 .avro;%0A%0A
+import com.fasterxml.jackson.annotation.JsonInclude;%0Aimport com.fasterxml.jackson.databind.MapperFeature;%0Aimport com.fasterxml.jackson.databind.ObjectMapper;%0Aimport com.fasterxml.jackson.databind.ObjectWriter;%0A
 import o
@@ -415,75 +415,8 @@
 ro;%0A
-import org.opencb.biodata.models.variant.avro.VariantFileMetadata;%0A
 impo
@@ -2330,369 +2330,406 @@
     
-// write variant metadata (header, stats,...) in a separated file%0A        FileOutputStream out = new FileOutputStream(metaFilename);%0A        AvroFileWriter%3CVariantFileMetadata%3E aw = new AvroFileWriter%3C%3E(VariantFileMetadata.SCHEMA$,%0A                                                                      compression, out);%0A        aw.open();%0A        aw.writeDatum
+ObjectMapper mapper = new ObjectMapper();%0A        mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);%0A        mapper.configure(MapperFeature.REQUIRE_SETTERS_FOR_GETTERS, true);%0A%0A        ObjectWriter writer = mapper.writer();%0A        PrintWriter pwriter = new PrintWriter(new FileWriter(metaFilename + %22.json%22));%0A        pwriter.write(writer.withDefaultPrettyPrinter().writeValueAsString
 (var
@@ -2753,27 +2753,8 @@
 l())
-;%0A        aw.close(
 );%0A 
@@ -2760,19 +2760,23 @@
         
-out
+pwriter
 .close()
@@ -2774,24 +2774,25 @@
 r.close();%0A%0A
+%0A
         // c

save variant metadata in Json format in VariantAvroSerializer; closes #46, closes #43