/********************************************************************************
 * Copyright (c) 2015-2018 Contributors to the Eclipse Foundation
 *
 * See the NOTICE file(s) distributed with this work for additional
 * information regarding copyright ownership.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0.
 *
 * SPDX-License-Identifier: EPL-2.0
 *
 ********************************************************************************/
package org.eclipse.mdm.businessobjects.utils;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;

import org.eclipse.mdm.protobuf.Mdm;
import org.eclipse.mdm.protobuf.Mdm.DateArray;
import org.eclipse.mdm.protobuf.Mdm.DoubleArray;
import org.eclipse.mdm.protobuf.Mdm.MeasuredValues;
import org.eclipse.mdm.protobuf.Mdm.MeasuredValues.Builder;
import org.eclipse.mdm.protobuf.Mdm.PreviewValuesList;
import org.eclipse.mdm.protobuf.Mdm.ScalarType;

import com.google.common.primitives.Bytes;
import com.google.common.primitives.Ints;
import com.google.protobuf.Timestamp;

/**
 * Helper class to calculate preview values.
 *
 */
public class PreviewHelper {
	/**
	 * Calcluates the preview values for numerical datatypes. For non numberical
	 * datatypes an empty MeasuredValues object is returned.
	 * 
	 * @param measuredValues List of measuredValues containing the mass data for the
	 *                       preview.
	 * @param numberOfChunks number of requested chunks for the preview
	 * @return calculated preview values
	 */
	public PreviewValuesList calculatePreview(List<MeasuredValues> measuredValues, int numberOfChunks) {
		return measuredValues.stream().map(mv -> calculatePreview(mv, numberOfChunks))
				.reduce(Mdm.PreviewValuesList.newBuilder(), (pl1, pl2) -> {
					pl1.addAllMin(pl2.getMinList());
					pl1.addAllMax(pl2.getMaxList());
					pl1.addAllAvg(pl2.getAvgList());
					return pl1;
				}).build();
	}

	/**
	 * Calculate preview for one {@link MeasuredValues}
	 * 
	 * @param measuredValues measured values
	 * @param numberOfChunks number of chunks
	 * @return a builder with the preview values set
	 */
	protected Mdm.PreviewValuesList.Builder calculatePreview(Mdm.MeasuredValues measuredValues, int numberOfChunks) {
		Mdm.MeasuredValues.Builder mv = Mdm.MeasuredValues.newBuilder(measuredValues).setScalarType(ScalarType.DOUBLE);

		switch (mv.getValuesCase()) {
		case BOOLEAN_ARRAY:
			return createEmpty(mv.clearBooleanArray());
		case BYTE_ARRAY:
			byte[] bytes = measuredValues.getByteArray().getValues().toByteArray();
			return calculate(mv, numberOfChunks, Bytes.asList(bytes), measuredValues.getFlagsList());
		case BYTE_STREAM_ARRAY:
			return createEmpty(mv.clearByteStreamArray());
		case DATE_ARRAY:
			List<Timestamp> timestamps = measuredValues.getDateArray().getValuesList();
			return calculateDate(mv, numberOfChunks, timestamps, measuredValues.getFlagsList());
		case DOUBLE_ARRAY:
			return calculate(mv, numberOfChunks, measuredValues.getDoubleArray().getValuesList(),
					measuredValues.getFlagsList());
		case DOUBLE_COMPLEX_ARRAY:
			return createEmpty(mv.clearDoubleComplexArray());
		case FLOAT_ARRAY:
			return calculate(mv, numberOfChunks, measuredValues.getFloatArray().getValuesList(),
					measuredValues.getFlagsList());
		case FLOAT_COMPLEX_ARRAY:
			return createEmpty(mv.clearFloatComplexArray());
		case INTEGER_ARRAY:
			return calculate(mv, numberOfChunks, measuredValues.getIntegerArray().getValuesList(),
					measuredValues.getFlagsList());
		case LONG_ARRAY:
			return calculate(mv, numberOfChunks, measuredValues.getLongArray().getValuesList(),
					measuredValues.getFlagsList());
		case SHORT_ARRAY:
			return calculate(mv, numberOfChunks, measuredValues.getShortArray().getValuesList(),
					measuredValues.getFlagsList());
		case STRING_ARRAY:
			return createEmpty(mv.clearStringArray());
		case VALUES_NOT_SET:
			throw new PreviewException(
					"Property 'values' of MeasuredValues with name '" + measuredValues.getName() + "' not set.");
		default:
			throw new PreviewException(
					"The value '" + mv.getValuesCase().name() + "' for Property 'values' of MeasuredValues with name '"
							+ measuredValues.getName() + "' is not supported.");
		}
	}

	/**
	 * Returns an empty builder.
	 * 
	 * @param mv
	 * @return a builder with no values or flags set
	 */
	private Mdm.PreviewValuesList.Builder createEmpty(Builder mv) {
		return Mdm.PreviewValuesList.newBuilder().addMin(mv.clearFlags().clearLength())
				.addMax(mv.clearFlags().clearLength()).addAvg(mv.clearFlags().clearLength());
	}

	/**
	 * Calculates min, max and avg for the given values for all data types except
	 * DT_DATE (@see this{@link #calculateDate(Builder, int, List, List)})
	 * 
	 * @param builder                 a builder to apply the values to.
	 * @param requestedNumberOfChunks number of chunks to calculate
	 * @param values                  measurement values
	 * @param flags                   list with flags
	 * @return an builder with the calculated values.
	 */
	private Mdm.PreviewValuesList.Builder calculate(Mdm.MeasuredValues.Builder builder, int requestedNumberOfChunks,
			List<? extends Number> values, List<Boolean> flags) {

		if (requestedNumberOfChunks < 1) {
			throw new PreviewException("Number of chunks requested must be positive!");
		}

		// numberOfChunks cannot be larger than length of data
		int numberOfChunks = Math.min(requestedNumberOfChunks, values.size());

		double chunkSize = values.size() / (double) numberOfChunks;

		DoubleArray.Builder minValues = DoubleArray.newBuilder();
		DoubleArray.Builder avgValues = DoubleArray.newBuilder();
		DoubleArray.Builder maxValues = DoubleArray.newBuilder();

		for (int chunkIndex = 0; chunkIndex < numberOfChunks; chunkIndex++) {
			double min = Double.MAX_VALUE;
			double sum = 0;
			double max = Double.MIN_VALUE;
			int count = 0;

			int startIndex = Ints.checkedCast(Math.round(chunkIndex * chunkSize));
			int endIndex = Ints.checkedCast(Math.round((chunkIndex + 1) * chunkSize));

			for (int valueIndex = startIndex; valueIndex < endIndex; valueIndex++) {
				if (valueIndex >= flags.size() || flags.get(valueIndex)) {
					double value = values.get(valueIndex).doubleValue();
					sum += value;
					min = Math.min(min, value);
					max = Math.max(max, value);
					count++;
				}
			}

			minValues.addValues(min);
			avgValues.addValues(sum / count);
			maxValues.addValues(max);
		}

		builder = builder.setLength(numberOfChunks);
		return PreviewValuesList.newBuilder().addMin(builder.setDoubleArray(minValues))
				.addAvg(builder.setDoubleArray(avgValues)).addMax(builder.setDoubleArray(maxValues));
	}

	/**
	 * Calculates min, max and avg for the given values for all data type DT_DATE.
	 * 
	 * @param builder                 a builder to apply the values to.
	 * @param requestedNumberOfChunks number of chunks to calculate
	 * @param values                  measurement values
	 * @param flags                   list with flags
	 * @return an builder with the calculated values.
	 */
	private Mdm.PreviewValuesList.Builder calculateDate(Mdm.MeasuredValues.Builder builder, int numberOfChunks,
			List<Timestamp> timestamps, List<Boolean> flags) {

		if (numberOfChunks < 1) {
			throw new PreviewException("Number of chunks requested must be positive!");
		}

		// numberOfChunks cannot be larger than length of data
		numberOfChunks = Math.min(numberOfChunks, timestamps.size());

		double chunkSize = timestamps.size() / (double) numberOfChunks;

		DateArray.Builder minValues = DateArray.newBuilder();
		DateArray.Builder avgValues = DateArray.newBuilder();
		DateArray.Builder maxValues = DateArray.newBuilder();

		for (int chunkIndex = 0; chunkIndex < numberOfChunks; chunkIndex++) {

			double sumSeconds = 0.0;
			double sumNanos = 0.0;

			Timestamp max = Timestamp.newBuilder().setSeconds(0L).setNanos(0).build();
			Timestamp min = Timestamp.newBuilder().setSeconds(Long.MAX_VALUE).setNanos(Integer.MAX_VALUE).build();
			int count = 0;

			int startIndex = Ints.checkedCast(Math.round(chunkIndex * chunkSize));
			int endIndex = Ints.checkedCast(Math.round((chunkIndex + 1) * chunkSize));

			for (int valueIndex = startIndex; valueIndex < endIndex; valueIndex++) {
				if (valueIndex >= flags.size() || flags.get(valueIndex)) {
					Timestamp value = timestamps.get(valueIndex);
					sumSeconds += value.getSeconds();
					sumNanos += value.getNanos();
					min = min(min, value);
					max = max(max, value);
					count++;
				}
			}
			minValues.addValues(min);
			maxValues.addValues(max);

			BigDecimal c = BigDecimal.valueOf(count);
			// calculate average and remainder
			BigDecimal[] avg = BigDecimal.valueOf(sumSeconds).add(BigDecimal.valueOf(sumNanos * 1E-9))
					.divideAndRemainder(c);

			// the remainder will be converted to nanoseconds
			avgValues.addValues(Timestamp.newBuilder().setSeconds(avg[0].longValue())
					.setNanos(avg[1].multiply(BigDecimal.valueOf(1E9)).divide(c, RoundingMode.HALF_UP).intValue()));
		}

		return PreviewValuesList.newBuilder().addMin(builder.setDateArray(minValues))
				.addAvg(builder.setDateArray(avgValues)).addMax(builder.setDateArray(maxValues));
	}

	/**
	 * Returns the smaller of the given timestamps.
	 * 
	 * @param t1 first timestamp
	 * @param t2 second timestamp
	 * @return t1, if t1 before t2 else t2
	 */
	private Timestamp min(Timestamp t1, Timestamp t2) {
		if (t1.getSeconds() < t2.getSeconds()) {
			return t1;
		} else if (t1.getSeconds() > t2.getSeconds()) {
			return t2;
		} else {
			return t1.getNanos() <= t2.getNanos() ? t1 : t2;
		}
	}

	/**
	 * Returns the larger of the given timestamps.
	 * 
	 * @param t1 first timestamp
	 * @param t2 second timestamp
	 * @return t1, if t1 after t2 else t2
	 */
	private Timestamp max(Timestamp t1, Timestamp t2) {
		if (t1.getSeconds() > t2.getSeconds()) {
			return t1;
		} else if (t1.getSeconds() < t2.getSeconds()) {
			return t2;
		} else {
			return t1.getNanos() >= t2.getNanos() ? t1 : t2;
		}
	}
}