package water.util;

import java.io.*;
import java.net.URI;
import java.util.Random;

import hex.FrameSplitter;
import jsr166y.CountedCompleter;
import water.*;
import water.fvec.Chunk;
import water.fvec.Frame;
import water.fvec.NFSFileVec;
import water.parser.ParseDataset;

public class FrameUtils {


  /** Parse given file(s) into the form of single frame represented by the given key.
   *
   * @param okey  destination key for parsed frame
   * @param files  files to parse
   * @return a new frame
   */
  public static Frame parseFrame(Key okey, File ...files) throws IOException {
    if (files == null || files.length == 0) {
      throw new IllegalArgumentException("List of files is empty!");
    }
    for (File f : files) {
      if (!f.exists())
        throw new FileNotFoundException("File not found " + f);
    }
    // Create output key if it is not given
    if(okey == null) okey = Key.make(files[0].getName());
    Key[] inKeys = new Key[files.length];
    for (int i=0; i<files.length; i++) inKeys[i] =  NFSFileVec.make(files[i])._key;
    return ParseDataset.parse(okey, inKeys);
  }

  public static Frame parseFrame(Key okey, URI ...uris) throws IOException {
    if (uris == null || uris.length == 0) {
      throw new IllegalArgumentException("List of uris is empty!");
    }
    Key[] inKeys = new Key[uris.length];
    for (int i=0; i<uris.length; i++)  inKeys[i] = H2O.getPM().anyURIToKey(uris[i]);
    if(okey == null) okey = Key.make(uris[0].toString());
    return ParseDataset.parse(okey, inKeys);
  }

  /**
   * Compute a chunk summary (how many chunks of each type, relative size, total size)
   * @param fr
   * @return chunk summary
   */
  public static ChunkSummary chunkSummary(Frame fr) {
    return new ChunkSummary().doAll(fr);
  }

  /** Generate given numbers of keys by suffixing key by given numbered suffix. */
  public static Key[] generateNumKeys(Key mk, int num) { return generateNumKeys(mk, num, "_part"); }
  public static Key[] generateNumKeys(Key mk, int num, String delim) {
    Key[] ks = new Key[num];
    String n = mk!=null ? mk.toString() : "noname";
    String suffix = "";
    if (n.endsWith(".hex")) {
      n = n.substring(0, n.length()-4); // be nice
      suffix = ".hex";
    }
    for (int i=0; i<num; i++) ks[i] = Key.make(n+delim+i+suffix);
    return ks;
  }

  /**
   * Helper to insert missing values into a Frame
   */
  public static class MissingInserter extends Job<Frame> {
    final Key _dataset;
    final double _fraction;
    final long _seed;

    public MissingInserter(Key frame, long seed, double frac){
      super(frame, "Missing Value Inserter");
      _dataset = frame; _seed = seed; _fraction = frac;
    }

    class MI extends MRTask<MI> {
      long _seed;
      double _frac;
      MI(long seed, double frac) {
        _seed=seed;
        _frac=frac;
      }
      @Override public void map (Chunk[]cs){
        final Random rng = new Random();
        for (int c = 0; c < cs.length; c++) {
          for (int r = 0; r < cs[c]._len; r++) {
            rng.setSeed(_seed + 1234 * c ^ 1723 * (cs[c].start() + r));
            if (rng.nextDouble() < _frac) cs[c].setNA(r);
          }
        }
        update(1);
      }
    }
    public void execImpl() {
      if (DKV.get(_dataset) == null)
        throw new IllegalArgumentException("Invalid Frame key " + _dataset + " (Frame doesn't exist).");
      if (_fraction < 0 || _fraction > 1 ) throw new IllegalArgumentException("fraction must be between 0 and 1.");

      Frame frame = DKV.getGet(_dataset);
      DKV.put(_progressKey = Key.make(), new Progress(frame.vecs()[0].nChunks()));
      new MI(_seed, _fraction).doAll(frame);
    }
  }

  /**
   * compute fraction of sparse chunks in this array.
   * @param chks
   * @return
   */
  public static double sparseRatio(Chunk [] chks) {
    int cnt = 0;
    double reg = 1.0/chks.length;
    for(Chunk c :chks)
      if(c.isSparse())
        ++cnt;
    return cnt * reg;
  }
}

@@ -2618,15 +2618,13 @@
 sing
- 
 Value
- 
 Inse
@@ -2738,120 +2738,75 @@
     
-long _seed;%0A      double _frac;%0A      MI(long seed, double frac) %7B%0A        _seed=seed;%0A        _frac=frac;%0A     
+final MissingInserter _mi;%0A      MI(MissingInserter mi) %7B _mi = mi;
  %7D%0A 
@@ -3010,16 +3010,20 @@
 etSeed(_
+mi._
 seed + 1
@@ -3097,20 +3097,28 @@
 le() %3C _
-frac
+mi._fraction
 ) cs%5Bc%5D.
@@ -3178,24 +3178,25 @@
     %7D%0A    %7D%0A
+%0A
     public v
@@ -3474,178 +3474,784 @@
 %22);%0A
-%0A      Frame frame = DKV.getGet(_dataset);%0A      DKV.put(_progressKey = Key.make(), new Progress(frame.vecs()%5B0%5D.nChunks()));%0A      new MI(_seed, _fraction).doAll(frame);
+      try %7B%0A        final MissingInserter mi = this;%0A        final Frame frame = DKV.getGet(_dataset);%0A        start(new H2O.H2OCountedCompleter() %7B%0A                @Override%0A                protected void compute2() %7B%0A                  new MI(mi).doAll(frame);%0A                  mi.done();%0A                %7D%0A                public boolean onExceptionalCompletion(Throwable ex, CountedCompleter cc) %7B%0A                  failed(ex);%0A                  return true;%0A                %7D%0A              %7D, frame.vecs()%5B0%5D.nChunks()%0A        );%0A      %7D catch (Throwable t) %7B%0A        Job thisJob = DKV.getGet(_key);%0A        if (thisJob._state == JobState.CANCELLED) %7B%0A          Log.info(%22Job cancelled by user.%22);%0A        %7D else %7B%0A          failed(t);%0A          throw t;%0A        %7D%0A      %7D
 %0A   

Make MissingInserter a proper Job with Progress bar and status.