#-------------------------------------------------------------
#
# 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.
#
#-------------------------------------------------------------
source("scripts/builtin/bandit.dml") as bandit;


# remove empty wrapper for frames
frameRemoveEmpty = function(Frame[Unknown] target, String marginParam, Matrix[Double] select)
return (Frame[Unknown] frameblock)
{
  idx = seq(1, ncol(target))
  # get the indexes of columns for recode transformation
  index = vectorToCsv(idx)
  # recode logical pipelines for easy handling
  jspecR = "{ids:true, recode:["+index+"]}";
  [Xd, M] = transformencode(target=target, spec=jspecR);
  X = replace(target=Xd, pattern = NaN, replacement=0)
  if(nrow(select) > 1 ) {
    # TODO fix removeEmpty Spark instruction to accept margin as a variable for now only support literal 
    if(marginParam == "rows")
      X = removeEmpty(target = X, margin = "rows", select = select)
    else
      X = removeEmpty(target = X, margin = "cols", select = select)
  }
  else { 
    if(marginParam == "rows")
      X = removeEmpty(target = X, margin = "rows")
    else
      X = removeEmpty(target = X, margin = "cols")
  }
  frameblock = transformdecode(target = Xd, spec = jspecR, meta = M)
  frameblock = frameblock[1:nrow(X), 1:ncol(X)]
}


#######################################################################
# Function for group-wise/stratified sampling from all classes in labelled dataset
# Inputs: The input dataset X, Y  and  sampling ratio between 0 and 1
# Output: sample X and Y
#######################################################################
doSample = function(Matrix[Double] eX, Matrix[Double] eY, Double ratio, Boolean verbose = FALSE)
  return (Matrix[Double] sampledX, Matrix[Double] sampledY)
{
  MIN_SAMPLE = 1000
  sampledX = eX
  sampledY = eY
  ratio = ifelse(nrow(eY) > 200000, 0.6, ratio)
  sampled = floor(nrow(eX) * ratio)
  
  if(sampled > MIN_SAMPLE & ratio != 1.0)
  {
    sampleVec = sample(nrow(eX), sampled, FALSE, 23)
    P = table(seq(1, nrow(sampleVec)), sampleVec, nrow(sampleVec), nrow(eX))
    if((nrow(eY) > 1))  # for classification
    {
      sampledX = P %*% eX
      sampledY = P %*% eY
    }
    else if(nrow(eY) == 1) { # for clustering
      sampledX = P %*% eX
      sampledY = eY 
    }
  }
}

# #######################################################################
# # 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) {

  if(sum(mask) > 0)
  {
    X = replace(target=X, pattern=NaN, replacement=1)
    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);
  }
  else dX_train = X
}


#####################################
# The function will check if the pipeline have zero hyper-parameters
# then it should not use more resource iterations and should be executed once
######################################
isResourceOptimal = function(List[Unknown] param, Boolean verbose)
return(Boolean validForResources) 
{
  validForResources = FALSE

  count = 0
  for(i in 1:length(param))
  {
    hp = as.matrix(param[i])
    if(ncol(hp) > 4)
      count += 1
  }
  validForResources = count > 0
}



#####################################
# The function will apply a pipeline of string processing primitives on dirty data
######################################
stringProcessing = function(Frame[Unknown] data, Matrix[Double] mask, 
  Frame[String] schema, Boolean CorrectTypos, List[Unknown] ctx = list(prefix="--"))
return(Frame[Unknown] data, List[Unknown] distanceMatrix, List[Unknown] dictionary, Matrix[Double] dateColIdx)
{ 

  prefix = as.scalar(ctx["prefix"]);
  distanceMatrix = list()
  dictionary = list()
  # step 1 do the case transformations
  print(prefix+" convert strings to lower case");
  data = map(data, "x -> x.toLowerCase()")
  # step 2 fix invalid lengths
  # q0 = 0.05
  # q1 = 0.95
  # print(prefix+" fixing invalid lengths between "+q0+" and "+q1+" quantile");

  # [data, mask, qlow, qup] = fixInvalidLengths(data, mask, q0, q1)

  
    # # step 3 fix swap values
  # print(prefix+" value swap fixing");
  # data = valueSwap(data, schema)
  
  # step 3 drop invalid types
  print(prefix+" drop values with type mismatch");
  data = dropInvalidType(data, schema)


  # step 5 porter stemming on all features
  # print(prefix+" porter-stemming on all features");
  # data = map(data, "x -> PorterStemmer.stem(x)", 0)

  # step 6 typo correction  
  if(CorrectTypos)
  {
    print(prefix+" correct typos in strings");
    # fix the typos
    for(i in 1:ncol(schema))
      if(as.scalar(schema[1,i]) == "STRING") {
        [d, ft, dt, dm, fr] = correctTypos(data[, i], 0.2, 0.9, FALSE);
        data[, i] = d
        distanceMatrix = append(distanceMatrix, dm)
        dictionary = append(distanceMatrix, fr)
      }
  }
  # # step 7 convert date to decimal
  dateColIdx = as.matrix(0)
  isDate = map(data[1:10], "x -> UtilFunctions.isDateColumn(x)")
  isDate = replace(target = as.matrix(isDate), pattern = NaN, replacement = 0)
  isDate = (colMaxs(isDate)) & as.matrix(schema == frame("STRING", rows=1, cols=ncol(schema)))
  if(sum(isDate) > 0) {
    print(prefix+" changing date to timestamp")
    dateColIdx = removeEmpty(target = isDate * t(seq(1, ncol(isDate))), margin="cols")
    for(i in 1:ncol(dateColIdx))
    {
      idx = as.scalar(dateColIdx[i])
      data[, idx] = map(data[, idx], "x -> UtilFunctions.getTimestamp(x)", margin=2)
    }
  }
  # TODO add deduplication
  print(prefix+" deduplication via entity resolution");
  
}

#####################################
# The function will apply a pipeline of string processing primitives on dirty data
######################################
stringProcessingApply = function(Frame[Unknown] data, Matrix[Double] mask, Frame[String] schema, 
  Boolean CorrectTypos, List[Unknown] distanceMatrix, List[Unknown] dictionary, Matrix[Double] dateColIdx)
return(Frame[Unknown] data)
{ 

  # step 1 do the case transformations
  data = map(data, "x -> x.toLowerCase()")
  # step 2 fix invalid lengths

  # q0 = 0.05
  # q1 = 0.95

  # [data, mask, qlow, qup] = fixInvalidLengths(data, mask, q0, q1)

  # # step 3 fix swap values
  # data = valueSwap(data, schema)

  # step 3 drop invalid types
  data = dropInvalidType(data, schema)


  # step 5 porter stemming on all features
  # data = map(data, "x -> PorterStemmer.stem(x)", 0)

  
  # step 6 typo correction  
  if(CorrectTypos)
  {
    # fix the typos
    for(i in 1:ncol(schema))
      if(as.scalar(schema[1,i]) == "STRING") {
          d = correctTyposApply(data[, i], 0.2, 0.9, as.matrix(distanceMatrix[i]), as.frame(dictionary[i]));
          data[, i] = d
      }
  }
  # # step 7 convert date to decimal
  if(sum(dateColIdx) > 0) {
    for(i in 1:ncol(dateColIdx))
    {
      idx = as.scalar(dateColIdx[i])
      data[, idx] = map(data[, idx], "x -> UtilFunctions.getTimestamp(x)", margin=2)
    }
  }
}