/**
 * 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.
 */
package org.apache.apex.malhar.python.base.util;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

import jep.NDArray;

/***
 * A handy Kryo serializer class that can be used to serialize and deserialize a JEP NDArray instance. It is
 *  recommended that {@link NDimensionalArray} be used in lieu of this class. This is because NDArray is highly specific
 *   JEP data structure and will not give flexibility if the python engines are changed in the future.
 */
public class NDArrayKryoSerializer extends Serializer<NDArray>
{
  private static final short TRUE_AS_SHORTINT = 1;

  private static final short FALSE_AS_SHORTINT = 0;

  @Override
  public void setGenerics(Kryo kryo, Class[] generics)
  {
    super.setGenerics(kryo, generics);
  }

  @Override
  public void write(Kryo kryo, Output output, NDArray ndArray)
  {

    Object dataVal = ndArray.getData();
    if (dataVal == null) {
      return;
    }
    // NDArray throws an exception in constructor if not an array. So below value will never be null
    Class classNameForArrayType = dataVal.getClass().getComponentType(); // null if it is not an array
    int[] dimensions = ndArray.getDimensions();
    boolean signedFlag = ndArray.isUnsigned();
    int arraySizeInSingleDimension = 1;
    for (int aDimensionSize : dimensions) {
      arraySizeInSingleDimension = arraySizeInSingleDimension * aDimensionSize;
    }
    // write the single dimension length
    output.writeInt(arraySizeInSingleDimension);
    // write the dimension of the dimensions int array
    output.writeInt(dimensions.length);
    // next we write the dimensions array itself
    output.writeInts(dimensions);

    // next write the unsigned flag
    output.writeBoolean(signedFlag);

    // write the data type of the N-dimensional Array
    if (classNameForArrayType != null) {
      output.writeString(classNameForArrayType.getCanonicalName());
    } else {
      output.writeString(null);
    }

    // write the array contents
    if (dataVal != null) {
      switch (classNameForArrayType.getCanonicalName()) {
        case "float":
          output.writeFloats((float[])dataVal);
          break;
        case "int":
          output.writeInts((int[])dataVal);
          break;
        case "double":
          output.writeDoubles((double[])dataVal);
          break;
        case "long":
          output.writeLongs((long[])dataVal);
          break;
        case "short":
          output.writeShorts((short[])dataVal);
          break;
        case "byte":
          output.writeBytes((byte[])dataVal);
          break;
        case "boolean":
          boolean[] originalBoolArray = (boolean[])dataVal;
          short[] convertedBoolArray = new short[originalBoolArray.length];
          for (int i = 0; i < originalBoolArray.length; i++) {
            if (originalBoolArray[i]) {
              convertedBoolArray[i] = TRUE_AS_SHORTINT;
            } else {
              convertedBoolArray[i] = FALSE_AS_SHORTINT;
            }
          }
          output.writeShorts(convertedBoolArray);
          break;
        default:
          throw new RuntimeException("Unsupported NDArray type serialization object");
      }
    }
  }

  @Override
  public NDArray read(Kryo kryo, Input input, Class<NDArray> aClass)
  {
    int singleDimensionArrayLength = input.readInt();
    int lengthOfDimensionsArray = input.readInt();
    int[] dimensions = input.readInts(lengthOfDimensionsArray);
    boolean signedFlag = input.readBoolean();

    String dataType = input.readString();
    if ( dataType == null) {
      return null;
    }
    switch (dataType) {
      case "float":
        NDArray<float[]> floatNDArray = new NDArray<>(
            input.readFloats(singleDimensionArrayLength),signedFlag,dimensions);
        return floatNDArray;
      case "int":
        NDArray<int[]> intNDArray = new NDArray<>(
            input.readInts(singleDimensionArrayLength),signedFlag,dimensions);
        return intNDArray;
      case "double":
        NDArray<double[]> doubleNDArray = new NDArray<>(
            input.readDoubles(singleDimensionArrayLength),signedFlag,dimensions);
        return doubleNDArray;
      case "long":
        NDArray<long[]> longNDArray = new NDArray<>(
            input.readLongs(singleDimensionArrayLength),signedFlag,dimensions);
        return longNDArray;
      case "short":
        NDArray<short[]> shortNDArray = new NDArray<>(
            input.readShorts(singleDimensionArrayLength),signedFlag,dimensions);
        return shortNDArray;
      case "byte":
        NDArray<byte[]> byteNDArray = new NDArray<>(
            input.readBytes(singleDimensionArrayLength),signedFlag,dimensions);
        return byteNDArray;
      case "boolean":
        short[] shortsArray = input.readShorts(singleDimensionArrayLength);
        boolean[] boolsArray = new boolean[shortsArray.length];
        for (int i = 0; i < shortsArray.length; i++) {
          if (TRUE_AS_SHORTINT == shortsArray[i]) {
            boolsArray[i] = true;
          } else {
            boolsArray[i] = false;
          }
        }
        NDArray<boolean[]> booleanNDArray = new NDArray<>(boolsArray,signedFlag,dimensions);
        return booleanNDArray;
      default:
        return null;
    }
  }
}