package com.tibbo.aggregate.common.expression.function;

import java.text.MessageFormat;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;

import javax.annotation.Nullable;

import org.apache.commons.math3.stat.descriptive.AbstractStorelessUnivariateStatistic;

import com.tibbo.aggregate.common.datatable.DataRecord;
import com.tibbo.aggregate.common.datatable.DataTable;
import com.tibbo.aggregate.common.datatable.FieldFormat;
import com.tibbo.aggregate.common.expression.EvaluationEnvironment;
import com.tibbo.aggregate.common.expression.EvaluationException;
import com.tibbo.aggregate.common.expression.Evaluator;

public abstract class AbstractSingleValueCollectorFunction extends AbstractFunction
{
  protected final AbstractStorelessUnivariateStatistic collector;
  private final ReentrantLock collectorLock = new ReentrantLock();
  
  /**
   * Note: If {@code collector} is null, developer should override {@code collect(DataTable table, FieldFormat ff)}, otherwise current implementation will throw {@code EvaluationException}.
   *
   * @see com.tibbo.aggregate.common.expression.function.math.ModeFunction
   */
  public AbstractSingleValueCollectorFunction(@Nullable AbstractStorelessUnivariateStatistic collector, String name, String category, String parametersFootprint, String returnValue,
      String description)
  {
    super(name, category, parametersFootprint, returnValue, description);
    this.collector = collector;
  }
  
  protected Object compare(Number first, Number second)
  {
    throw new RuntimeException(MessageFormat.format("Comparison of 2 arguments is not implemented for \"{0}\" function",
        getName()));
  }
  
  @Override
  public Object execute(Evaluator evaluator, EvaluationEnvironment environment, Object... parameters) throws EvaluationException
  {
    if (parameters.length == 2 && parameters[0] instanceof Number && parameters[1] instanceof Number)
    {
      return compare((Number) parameters[0], (Number) parameters[1]);
    }
    
    checkParameters(1, false, parameters);
    checkParameterType(0, parameters[0], DataTable.class);
    
    DataTable table = (DataTable) parameters[0];
    if (table.getRecordCount() == 0)
    {
      return 0.0d;
    }
    
    FieldFormat ff = table.getFormat(0);
    if (parameters.length >= 2)
    {
      ff = checkAndGetNumericTypeField(table, parameters[1]);
    }
    else
    {
      checkNumericTypeField(ff);
    }
    
    return collect(table, ff);
  }
  
  protected Object collect(DataTable table, FieldFormat ff) throws EvaluationException
  {
    if (collector == null)
    {
      throw new EvaluationException("No collector is set");
    }
    
    collectorLock.lock();
    try
    {
      collector.clear();
      Function<Number, Double> toDoubleConverter = getDoubleConvert(ff.getType());
      
      String fieldName = ff.getName();
      table.forEach(rec -> {
        Object value = rec.getValue(fieldName);
        if (value != null)
        {
          collector.increment(toDoubleConverter.apply((Number) value));
        }
      });
      
      return collector.getResult();
    }
    finally
    {
      collectorLock.unlock();
    }
  }
  
  private static Function<Number, Double> getDoubleConvert(char entityType)
  {
    switch (entityType)
    {
      case FieldFormat.INTEGER_FIELD:
        return (value) -> (double) value.intValue();
      case FieldFormat.LONG_FIELD:
        return (value) -> (double) value.longValue();
      case FieldFormat.FLOAT_FIELD:
        return (value) -> Double.valueOf(value.toString());
      case FieldFormat.DOUBLE_FIELD:
        return (value) -> value.doubleValue();
      default:
        throw new IllegalStateException("Expected numeric type, got: " + entityType);
    }
  }
  
  protected double[] toArray(DataTable table, FieldFormat ff)
  {
    String field = ff.getName();
    double[] data = new double[table.getRecordCount()];
    int i = 0;
    
    Function<Number, Double> toDoubleConverter = getDoubleConvert(ff.getType());
    
    for (DataRecord rec : table)
    {
      Object value = rec.getValue(field);
      if (value != null)
      {
        data[i++] = toDoubleConverter.apply((Number) value);
      }
    }
    return data;
  }
}
