#-------------------------------------------------------------
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.
#
#-------------------------------------------------------------

s_executePipeline = function(Frame[String] pipeline, Matrix[Double] X,  Matrix[Double] Y, Matrix[Double] mask,
  Matrix[Double] FD, Matrix[Double] hyperParameters, Integer flagsCount, Boolean verbose)
  return (Matrix[Double] X, Matrix[Double] Y, Double t2)
{
  t1 = time();
  print("PIPELINE EXECUTION START ... "+toString(pipeline))

  if(verbose) {
    print("checks   rows in X = "+nrow(X)+" rows in Y = "+nrow(Y)+" cols in X = "+ncol(X)+" col in Y = "+ncol(Y))
    print("pipeline in execution "+toString(pipeline))
    print("pipeline hps "+toString(hyperParameters))
    while(FALSE){}
  }
  for(i in 1:ncol(pipeline)) {
    op = as.scalar(pipeline[1,i])
    [hp, withClass, dataFlag] = matrixToList(X, Y, mask, FD, hyperParameters[i], flagsCount, op)
    Xclone = X
    X = eval(op, hp)
    # dataFlag 0 = only on numeric, 1 = on whole data
    X = confirmData(X, Xclone, mask, dataFlag)
    if(withClass)
    {
      Y = X[, ncol(X)]
      X = X[, 1:ncol(X) - 1]
    }

    X = confirmMeta(X, mask)
  }
  t2 = floor((time() - t1) / 1e+6)
  print("PIPELINE EXECUTION ENDED: "+t2+" ms")
}

# This function will convert the matrix row-vector into list
matrixToList = function(Matrix[Double] X,  Matrix[Double] Y, Matrix[Double] mask, Matrix[Double] FD,
  Matrix[Double] p, Integer flagsCount, String op)
  return (List[Unknown] l, Boolean hasY, Integer dataFlag)
{
  NUM_META_FLAGS = flagsCount;
  hasY = FALSE

  dataFlag = as.integer(as.scalar(p[1, ncol(p)]))
  hasVerbose = as.scalar(p[1, ncol(p) - 1])
  yFlag = as.scalar(p[1, ncol(p) - 2])
  fDFlag = as.integer(as.scalar(p[1, ncol(p)-3]))
  maskFlag = as.integer(as.scalar(p[1, ncol(p)-4]))
  
  ######################################################
  # CHECK FOR DATA FLAG
  if(dataFlag == 0)
  { 
    # take numerics out
    X = removeEmpty(target=X, margin = "cols", select = (mask == 0))
  }
  else if(dataFlag == 1)
  { 
    # take categorical out
    X = removeEmpty(target=X, margin = "cols", select = mask)
    # print("data for execution \n"+toString(X, rows=5))
  } 
  
  l = list(X)
    
  ######################################################
  # CHECK FOR Y APPEND FLAG  
 
  if(yFlag == 1) {
    l = append(l, Y)
    hasY = TRUE
  }
  ######################################################
  # CHECK FOR FD APPEND FLAG
  if(fDFlag == 1)
  {
    l = append(l, FD)
  }
  
  ######################################################
  # CHECK FOR MASK APPEND FLAG
  if(maskFlag == 1)
  {
    l = append(l, mask)
  }
  #####################################################
  # POPULATE HYPER PARAM
  # get the number of hyper-parameters and loop till that
  no_of_hyperparam = as.scalar(p[1,1])
  if(no_of_hyperparam > 0) {
    # if(op == "pca") {
      # # convert the number parameters to a ration related to OHE columns
      # ratio = as.scalar(p[resource_index,1])
      # p[resource_index, 1] = as.integer(ncol(X) - ratio)
    # }
    for(i in 1:no_of_hyperparam)
      l = append(l, as.scalar(p[1,(i+1)]))
  }
  ######################################################
  # CHECK FOR VERBOSE FLAG
  if(hasVerbose == 1)
    l = append(l, FALSE)

}

confirmMeta = function(Matrix[Double] X, Matrix[Double] mask)
return (Matrix[Double] X)
{
  if((sum(mask) > 0) & (ncol(X) == ncol(mask)))
  {
    # get  the max + 1 for nan replacement
    nanMask = is.na(X)
    # replace nan
    X = replace(target = X, pattern = NaN, replacement = 9999)
    # take categorical out
    cat = removeEmpty(target=X, margin="cols", select = mask)
    # round categorical (if there is any floating  point)
    cat = ceil(cat)
    # reconstruct original X
    X = X * (mask == 0)
    q = table(seq(1, ncol(cat)), removeEmpty(target=seq(1, ncol(mask)), margin="rows", 
      select=t(mask)), ncol(cat), ncol(X))
    X = (cat %*% q) + X 

    # put nan back
    nanMask = replace(target = nanMask, pattern = 1, replacement = NaN)
    X = X + nanMask
    # print("X less than equal to  zero "+sum(cat <= 0))
  }
}


confirmData = function(Matrix[Double] nX, Matrix[Double] originalX, Matrix[Double] mask, Integer dataFlag)
return (Matrix[Double] X)
{
  # print("changes data \n"+toString(nX, rows=10))
  
  while(FALSE){}
  if(dataFlag == 0 & (sum(mask) > 0))
  {
    maxDummy = max(nX) + 1
    nX = replace(target = nX, pattern = NaN, replacement = maxDummy)
    # X without numerics
    Xcat = removeEmpty(target=originalX, margin="cols", select=mask)
    nanMask = is.na(Xcat)
    Xcat = replace(target = Xcat, pattern = NaN, replacement = -1111)
    # print("unchanged data \n"+toString(originalX, rows=10))
    
    # reconstruct the original matrix
    p = table(seq(1, ncol(nX)), removeEmpty(target=seq(1, ncol(mask)), margin="rows", 
    select=t(mask==0)), ncol(nX), ncol(originalX))
    q = table(seq(1, ncol(Xcat)), removeEmpty(target=seq(1, ncol(mask)), margin="rows", 
    select=t(mask)), ncol(Xcat), ncol(originalX))
    X = (nX %*% p) + (Xcat %*% q) 

    X = replace(target = X, pattern = maxDummy, replacement = NaN)
    X = replace(target = X, pattern = -1111, replacement = NaN)
  }
  else if(dataFlag == 1 & (sum(mask) > 0))
  {
    maxDummy = max(nX) + 1
    nX = replace(target = nX, pattern = NaN, replacement = maxDummy)
    # X without categorical
    Xnum = removeEmpty(target=originalX, margin="cols", select=(mask==0))
    nanMask = is.na(Xnum)
    Xnum = replace(target = Xnum, pattern = NaN, replacement = -1111)
    # reconstruct the original matrix
    p = table(seq(1, ncol(Xnum)), removeEmpty(target=seq(1, ncol(mask)), margin="rows", 
    select=t(mask==0)), ncol(Xnum), ncol(originalX))
    q = table(seq(1, ncol(nX)), removeEmpty(target=seq(1, ncol(mask)), margin="rows", 
    select=t(mask)), ncol(nX), ncol(originalX))
    X = (nX %*% q) + (Xnum %*% p) 
    X = replace(target = X, pattern = maxDummy, replacement = NaN)
    X = replace(target = X, pattern = -1111, replacement = NaN)
  
  }
  else X = nX
    # print("recreated data \n"+toString(X, rows = 20))
}


#######################################################################
# Wrapper of transformencode OHE call, to call inside eval as a function
# Inputs: The input dataset X, and  mask of the columns
# Output: OHEd matrix X
#######################################################################

dummycoding = function(Matrix[Double] X, Matrix[Double] mask)
return (Matrix[Double] dX_train) {
  X = replace(target=X, pattern=NaN, replacement=0)
  idx = vectorToCsv(mask)
  
  # specifications for one-hot encoding of categorical features
  jspecDC = "{ids:true, dummycode:["+idx+"]}";
  # OHE of categorical features
  [dX_train, dM] = transformencode(target=as.frame(X), spec=jspecDC);

}


#######################################################################
# Wrapper of imputeByFD OHE call, to call inside eval as a function
# Inputs: The input dataset X, and  mask of the columns and threshold value
# Output: filled matrix X
#######################################################################

imputeByFd = function(Matrix[Double] X, Matrix[Double] FD,  Double threshold)
return (Matrix[Double] X_filled)
{
  if(sum(FD) > 0)
  {
    for(i in 1: nrow(FD))
    {
      for(j in 1:ncol(FD)) {
        if(as.scalar(FD[i, j]) > 0 & (min(X[, i]) != 0) & (min(X[, j]) != 0) & (sum(FD[, j]) != nrow(FD)))
          X = imputeByFD(X, i, j, threshold, FALSE)
      }
    }
  }
  X_filled = X
}

#######################################################################
# Wrapper of na_lof to call inside eval as a function
# Output: filled matrix X
#######################################################################

forward_fill = function(Matrix[Double] X, Boolean op, Boolean verbose)
return (Matrix[Double] X_filled)
{ 
  option = ifelse(op, "locf", "nocb")
  X_filled = na_locf(X=X, option=option, verbose=verbose)
}



# smote wrapper for doing relative over-sampling
SMOTE  = function(Matrix[Double] X, Matrix[Double] Y, Matrix[Double] mask, Integer remainingRatio, Boolean verbose)
return (Matrix[Double] XY)
{
  XY = order(target = cbind(Y, X),  by = 1, decreasing=FALSE, index.return=FALSE)
  synthesized = matrix(0,0,0) # initialize variable
  # get the class count 
  classes = table(XY[, 1], 1)
  start_class = 1
  end_class = 0
  k = table(XY[, 1], 1)
  getMax = max(k)
  maxKIndex = as.scalar(rowIndexMax(t(k)))
  outSet = matrix(0, 0, ncol(XY))
  remainingRatio = ifelse((remainingRatio%%100) >= 50, remainingRatio+(100 - (remainingRatio%%100)),
  remainingRatio-(remainingRatio%%100))
  for(i in 1: nrow(k)) {
    end_class = end_class + as.scalar(classes[i])
    class_t = XY[start_class:end_class, ]
    if((i != maxKIndex)) {
      synthesized = smote(class_t[, 2:ncol(XY)], mask, remainingRatio, 1, FALSE)
      synthesized = cbind(matrix(as.scalar(class_t[2,1]), nrow(synthesized), 1), synthesized)
      outSet = rbind(outSet, synthesized)
    }
    start_class = end_class + 1
  }
  
  XY = rbind(XY, synthesized)
  Y = XY[, 1]
  X = XY[, 2:ncol(XY)]
  XY = cbind(X,Y)
  classes = table(Y, 1)
}


########################################################
# The function will replace the null with default values
########################################################
fillDefault = function(Matrix[Double] X)
return(Matrix[Double] X){
  defaullt = round(colMaxs(X) - colMins(X))
  Mask = is.na(X)
  X = replace(target=X, pattern=NaN, replacement=max(X))
  Mask = Mask * defaullt
  X = X + Mask
  print("fillDefault: no of NaNs "+sum(is.na(X)))
}

########################################################
# A slightly changes version of PCA
########################################################
m_pca = function(Matrix[Double] X, Integer K=2, Boolean center=TRUE, Boolean scale=TRUE)
  return (Matrix[Double] Xout) 
{
  if(K < ncol(X)) {
    N = nrow(X);
    D = ncol(X);

    # perform z-scoring (centering and scaling)
    [X, Centering, ScaleFactor] = scale(X, center, scale);

    # co-variance matrix
    mu = colSums(X)/N;
    C = (t(X) %*% X)/(N-1) - (N/(N-1))*t(mu) %*% mu;

    # compute eigen vectors and values
    [evalues, evectors] = eigen(C);

    decreasing_Idx = order(target=evalues,by=1,decreasing=TRUE,index.return=TRUE);
    diagmat = table(seq(1,D),decreasing_Idx);
    # sorts eigenvalues by decreasing order
    evalues = diagmat %*% evalues;
    # sorts eigenvectors column-wise in the order of decreasing eigenvalues
    evectors = evectors %*% diagmat;

    eval_dominant = evalues[1:K, 1];
    evec_dominant = evectors[,1:K];

    # Construct new data set by treating computed dominant eigenvectors as the basis vectors
    Xout = X %*% evec_dominant;
    Mout = evec_dominant;
  }
  else Xout = X

}



