package hex;


import water.*;
import water.ValueArray.Column;

public class ScoreTask extends MRTask {

  final Model _M;
  final ValueArray _outputAry;
  final int _nchunks;
  private double _min = Double.POSITIVE_INFINITY;
  private double _max = Double.NEGATIVE_INFINITY;
  long _n;
  private double _sum;
  private double _sumsq;

  private int [] _rpc;

  private ScoreTask(Model M,ValueArray output, int nchunks){_M = M; _outputAry=output; _nchunks = nchunks;}

  public static Key score(Model M, ValueArray data, Key outputKey){
    ValueArray output = new ValueArray(outputKey, data.length());
    ScoreTask t = new ScoreTask(M.adapt(data),output,(int)data.chunks());
    t.invoke(data._key);
    Column c = new Column();
    c._max = t._max;
    c._min = t._min;
    c._mean = t._sum/t._n;
    double norm = 1.0/t._n;
    double s = t._sum*norm;
    c._sigma =  Math.sqrt(t._sumsq*norm - s*s);
    c._name = M.responseName();
    c._size = -8;
    c._scale = 1;
    c._base = 0;
    c._off = 0;
    c._domain = null;
    ValueArray res = new ValueArray(outputKey,t._rpc,8, new Column[]{c});
    DKV.put(outputKey, res);
    return res._key;
  }

  @Override public void init(){
    _rpc = new int[_nchunks];
    super.init();
  }
  @Override public void map(Key key) {
    ValueArray ary = DKV.get(ValueArray.getArrayKey(key)).get();
    AutoBuffer bits = new AutoBuffer(DKV.get(key).memOrLoad());
    int nrows = bits.remaining()/ary._rowsize;
    AutoBuffer res = new AutoBuffer(nrows<<3);
    for(int i = 0; i< nrows; ++i){
      double p = _M.score(ary, bits, i);
      if(p < _min)_min = p;
      if(p > _max)_max = p;

      if(!Double.isNaN(p)){
        _sum += p;
        _sumsq += p*p;
        ++_n;
      }
      res.put8d(p);
    }
    int idx = (int)ValueArray.getChunkIndex(key);
    Key outputKey = _outputAry.getChunkKey(idx);
    _rpc[idx] += nrows;
    DKV.put(outputKey, new Value(outputKey,res.buf()));
  }

  @Override public void reduce(DRemoteTask drt) {
    ScoreTask t = (ScoreTask)drt;
    _min = Math.min(_min,t._min);
    _max = Math.max(_max,t._max);
    _sum += t._sum;
    _sumsq += t._sumsq;
    _n += t._n;
    if(_rpc == null)
      _rpc = t._rpc;
    else if(_rpc != t._rpc)
      for(int i = 0; i < _rpc.length; ++i)
        _rpc[i] += t._rpc[i];
  }
}

@@ -128,28 +128,18 @@
 nal 
-ValueArra
+Ke
 y _out
-putAr
+Ke
 y;%0A 
@@ -371,33 +371,27 @@
 Model M,
-ValueArra
+ Ke
 y out
-put
+Key
 , int nc
@@ -401,16 +401,21 @@
 ks)%7B
+%0A    
 _M = M; 
 _out
@@ -414,26 +414,34 @@
  M; 
-_outputAry=output;
+%0A    _outKey = outKey;%0A   
  _nc
@@ -456,16 +456,19 @@
 nchunks;
+%0A  
 %7D%0A%0A  pub
@@ -534,74 +534,8 @@
 y)%7B%0A
-    ValueArray output = new ValueArray(outputKey, data.length());%0A
     
@@ -582,16 +582,19 @@
 ),output
+Key
 ,(int)da
@@ -1757,17 +1757,17 @@
 y = 
-_outputAr
+ValueArra
 y.ge
@@ -1779,16 +1779,24 @@
 kKey(idx
+,_outKey
 );%0A    _

Fix ScoreTask NPE bug