/*
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.edgent.analytics.math3.json;

import java.util.List;

import org.apache.commons.math3.util.Pair;
import org.apache.edgent.analytics.math3.Aggregations;
import org.apache.edgent.function.BiFunction;
import org.apache.edgent.function.ToDoubleFunction;
import org.apache.edgent.topology.TStream;
import org.apache.edgent.topology.TWindow;

import com.google.gson.JsonElement;
import com.google.gson.JsonObject;

/**
 * Apache Common Math analytics for streams with JSON tuples.
 *
 * @see Aggregations
 */
public class JsonAnalytics {
    
    /**
     * Aggregate against a single {@code Numeric} variable contained in an JSON object.
     * 
     * The returned stream contains a tuple for each execution performed against a window partition.
     * The tuple is a {@code JsonObject} containing:
     * <UL>
     * <LI> Partition key of type {@code K} as a property with key {@code resultPartitionProperty}. </LI>
     * <LI> Aggregation results as a {@code JsonObject} as a property with key {@code valueProperty}.
     * This results object contains the results of all aggregations defined by {@code aggregates} against
     * {@code double} property with key {@code valueProperty}.
     * <BR>
     * Each {@link JsonUnivariateAggregate} declares how it represents its aggregation in this result
     * object.
     * </LI>
     * </UL>
     * <P>
     * For example if the window contains these three tuples (pseudo JSON) for
     * partition 3:
     * <BR>
     * <code>{id=3,reading=2.0}, {id=3,reading=2.6}, {id=3,reading=1.8}</code>
     * <BR>
     * the resulting aggregation for the stream returned by:
     * <BR>
     * {@code aggregate(window, "id", "reading", Statistic.MIN, Statistic.MAX)}
     * <BR>
     * would contain this tuple with the maximum and minimum values in the {@code reading}
     * JSON object:
     * <BR>
     * <code>{id=3, reading={MIN=1.8, MAX=1.8}}</code>
     * </P>
     * @param <K> Partition type
     * 
     * @param window Window to aggregate over.
     * @param resultPartitionProperty Property to store the partition key in tuples on the returned stream.
     * @param valueProperty JSON property containing the value to aggregate.
     * @param aggregates Which aggregations to be performed.
     * @return Stream that will contain aggregations.
     */
    public static <K extends JsonElement> TStream<JsonObject> aggregate(
            TWindow<JsonObject, K> window,
            String resultPartitionProperty,
            String valueProperty,
            JsonUnivariateAggregate... aggregates) {
        return aggregate(window, resultPartitionProperty, valueProperty, j -> j.get(valueProperty).getAsDouble(), aggregates);

    }
    
    /**
     * Aggregate against a single {@code Numeric} variable contained in an JSON object.
     * 
     * The returned stream contains a tuple for each execution performed against a window partition.
     * The tuple is a {@code JsonObject} containing:
     * <UL>
     * <LI> Partition key of type {@code K} as a property with key {@code resultPartitionProperty}. </LI>
     * <LI> Aggregation results as a {@code JsonObject} as a property with key {@code resultProperty}.
     * This results object contains the results of all aggregations defined by {@code aggregates} against
     * value returned by {@code valueGetter}.
     * <BR>
     * Each {@link JsonUnivariateAggregate} declares how it represents its aggregation in this result
     * object.
     * </LI>
     * </UL>
     * 
     * @param <K> Partition type
     * @param window Window to aggregate over.
     * @param resultPartitionProperty Property to store the partition key in tuples on the returned stream.
     * @param resultProperty Property to store the aggregations in tuples on the returned stream.
     * @param valueGetter How to obtain the single variable from input tuples.
     * @param aggregates Which aggregations to be performed.
     * @return Stream that will contain aggregations.
     */
    public static <K extends JsonElement> TStream<JsonObject> aggregate(
            TWindow<JsonObject, K> window,
            String resultPartitionProperty,
            String resultProperty,
            ToDoubleFunction<JsonObject> valueGetter,
            JsonUnivariateAggregate... aggregates) {

        return window.aggregate(aggregateList(
                resultPartitionProperty,
                resultProperty,
                valueGetter,
                aggregates
                ));
    }
    
    /**
     * Create a Function that aggregates against a single {@code Numeric}
     * variable contained in an JSON object.
     * 
     * Calling {@code apply(List<JsonObject>)} on the returned {@code BiFunction}
     * returns a {@link JsonObject} containing:
     * <UL>
     * <LI> Partition key of type {@code K} as a property with key {@code resultPartitionProperty}. </LI>
     * <LI> Aggregation results as a {@code JsonObject} as a property with key {@code valueProperty}.
     * This results object contains the results of all aggregations defined by {@code aggregates}
     * against the value returned by {@code valueGetter}.
     * <BR>
     * Each {@link JsonUnivariateAggregate} declares how it represents its aggregation in this result
     * object.
     * </LI>
     * </UL>
     * <P>
     * For example if the list contains these three tuples (pseudo JSON) for
     * partition 3:
     * <BR>
     * <code>{id=3,reading=2.0}, {id=3,reading=2.6}, {id=3,reading=1.8}</code>
     * <BR>
     * the resulting aggregation for the JsonObject returned by:
     * <BR>
     * {@code aggregateList("id", "reading", Statistic.MIN, Statistic.MAX).apply(list, 3)}
     * <BR>
     * would be this tuple with the maximum and minimum values in the {@code reading}
     * JSON object:
     * <BR>
     * <code>{id=3, reading={MIN=1.8, MAX=1.8}}</code>
     * </P>
     * @param <K> Partition type
     * 
     * @param resultPartitionProperty Property to store the partition key in tuples on the returned stream.
     * @param resultProperty Property to store the aggregations in the returned JsonObject.
     * @param valueGetter How to obtain the single variable from input tuples.
     * @param aggregates Which aggregations to be performed.
     * @return Function that performs the aggregations.
     */
    public static <K extends JsonElement> 
    BiFunction<List<JsonObject>, K, JsonObject> aggregateList(
            String resultPartitionProperty,
            String resultProperty,
            ToDoubleFunction<JsonObject> valueGetter,
            JsonUnivariateAggregate... aggregates) {

        BiFunction<List<JsonObject>, K, JsonObject> function = (tuples, partition) -> {
            
            final JsonUnivariateAggregator[] aggregators = new JsonUnivariateAggregator[aggregates.length];
            for (int i = 0; i < aggregates.length; i++) {
                aggregators[i] = aggregates[i].get();
            }     
            
            final JsonObject result = new JsonObject();
            result.add(resultPartitionProperty, partition);
            JsonObject aggregateResults = new JsonObject();
            result.add(resultProperty, aggregateResults);

            final int n = tuples.size();
            aggregateResults.addProperty(JsonUnivariateAggregate.N, n);
            
            if (n != 0) {

                for (JsonUnivariateAggregator agg : aggregators) {
                    agg.clear(partition, n);
                }
                for (JsonObject tuple : tuples) {
                    double v = valueGetter.applyAsDouble(tuple);
                    for (JsonUnivariateAggregator agg : aggregators) {
                        agg.increment(v);
                    }
                }
                for (JsonUnivariateAggregator agg : aggregators) {
                    agg.result(partition, aggregateResults);
                }
            }

            return result;
        };
        
        return function;
    }
    
    /**
     * Aggregate against multiple {@code Numeric} variables contained in an JSON object.
     * <P>
     * This is a multi-variable analog of {@link #aggregate(TWindow, String, String, JsonUnivariateAggregate...) aggregate()}
     * </P>
     * <P>
     * See {@link #mvAggregateList(String, String, List) mvAggregateList()} for
     * a description of the aggregation processing and result stream.
     * </P>
     * <P>
     * Sample use:
     * <pre>{@code
     * // Ingest the data.  The JsonObject tuples have properties:
     * //   "id" - the partitionKey
     * //   "tx" - a numeric data variable
     * //   "rx" - a numeric data variable
     * TStream<JsonObject> ingestData = ...
     * 
     * // Define the tuple variables and their aggregations to compute
     * List<Pair<String, JsonUnivariateAggregate[]>> aggSpecs = new ArrayList<>();
     * aggSpecs.add(mkAggregationSpec("tx", Statistics.MIN, Statistics.MAX));
     * aggSpecs.add(mkAggregationSpec("rx", Statistics.MEAN));
     * 
     * // Create the window over which to aggregate
     * TWindow<JsonObject, JsonElement> window = 
     *    ingestData.last(5, TimeUnit.SECONDS, jo -> jo.get("id"));
     * 
     * // Create a stream with the aggregations. The result tuples have properties:
     * //   "id" - the partitionKey
     * //   "aggResults" - the aggregation results
     * TStream<JsonObject> aggResults = 
     *    mvAggregate(window, "id", "aggResults", aggSpecs);
     *    
     * // Create a stream of JsonObject tuples with just the average "rx"
     * TStream<JsonObject> avgRx = aggResults.map(
     *    jo -> {
     *      JsonObject result = new JsonObject();
     *      result.add("id", jo.get("id"))
     *      result.add("avgRx", getMvAggregate(jo, "aggResults", "Rx", Statistic.MEAN);
     *      return result;
     *    });
     * }</pre>
     * 
     * @param window the window to compute aggregations over
     * @param resultPartitionKeyProperty name of the partition key property in the result
     * @param resultProperty name of the aggregation results property in the result
     * @param aggregateSpecs see {@link #mkAggregationSpec(String, JsonUnivariateAggregate...) mkAggregationSpec()}
     * @return TStream&lt;JsonObject&gt; with aggregation results
     * 
     * @see #mvAggregateList(String, String, List) mvAggregateList()
     * @see #mkAggregationSpec(String, JsonUnivariateAggregate...) mkAggregationSpec()
     * @see #getMvAggregate(JsonObject, String, String, JsonUnivariateAggregate) getMvAggregate()
     */
    public static <K extends JsonElement> TStream<JsonObject> mvAggregate(
        TWindow<JsonObject, K> window,
        String resultPartitionKeyProperty,
        String resultProperty,
        List<Pair<String, JsonUnivariateAggregate[]>> aggregateSpecs) {

      return window.aggregate(mvAggregateList(
              resultPartitionKeyProperty,
              resultProperty,
              aggregateSpecs
              ));
    }

    /**
     * Create an aggregation specification.
     * <P>
     * The aggregation specification specifies a variable name and
     * the aggregates to compute on it.
     * </P>
     * <P>
     * The specification can be use with {@link #mvAggregateList(String, String, List) mkAggregateList()}
     * 
     * @param variableName the name of a {@code Numeric} data variable in a JSON object 
     * @param aggregates the aggregates to compute for the variable
     * @return the aggregation specification
     */
    public static Pair<String, JsonUnivariateAggregate[]> 
    mkAggregationSpec(String variableName, JsonUnivariateAggregate... aggregates) {
      return new Pair<String, JsonUnivariateAggregate[]>(variableName, aggregates);
    }
    
    /**
     * Create a Function that aggregates multiple {@code Numeric}
     * variables contained in an JSON object.
     * <P>
     * This is a multi-variable analog of {@link JsonAnalytics#aggregateList(String, String, org.apache.edgent.function.ToDoubleFunction, JsonUnivariateAggregate...) aggregateList()}
     * <P>
     * The overall multi-variable aggregation result is a JSON object
     * with properties:
     * <ul>
     * <li>{@code resultPartionKeyProperty} whose value is the tuple's partition key
     * <li>{@code resultProperty} whose value is a JSON object containing
     *     a property for each variable aggregation.  The property names
     *     correspond to the variable names from the {@code aggregateSpecs}
     *     and the values are the aggregation results for the variable.
     *     The aggregation results for a variable are a JSON object 
     *     having a property for each aggregation name and its value.</li>
     * </ul>
     * <P>
     * For example if the list contains these three tuples (pseudo JSON) for
     * partition 3:
     * <BR>
     * <code>{id=3,tx=2.0,rx=1.0,...}, {id=3,tx=2.6,rx=2.0,...}, {id=3,tx=1.8,rx=3.0,...}</code>
     * <BR>
     * the resulting aggregation JsonObject returned is:
     * <BR>
     * <code>{id=3, aggData={tx={MIN=1.8, MAX=2.6}, rx={MEAN=2.0}}}</code>
     * <BR>
     * for the invocation:
     * <BR>
     * <code>mvAggregateList("id", "aggData", aggSpecs).apply(list, 3))</code>
     * <BR>
     * where {@code aggSpecs} is:
     * <BR>
     * {@code
     * aggSpecs.add(mkAggregationSpec("tx", Statistics.MIN, Statistics.MAX));
     * aggSpecs.add(mkAggregationSpec("rx", Statistics.MEAN));
     * }
     * </P>
     * <P>
     * {@link #getMvAggregate(JsonObject, String, String, JsonUnivariateAggregate) getMvAggregate()}
     * can be used to extract individual aggregate values from the result.
     * </P>
     * 
     * @param <K> Partition Key as a JsonElement
     * 
     * @param resultPartitionKeyProperty name of the partition key property in the result
     * @param resultProperty name of the aggregation results property in the result
     * @param aggregateSpecs see {@link #mkAggregationSpec(String, JsonUnivariateAggregate...) mkAggregationSpec()}
     * @return Function that performs the aggregations.
     * 
     * @see #mkAggregationSpec(String, JsonUnivariateAggregate...) mkAggregationSpec()
     * @see #getMvAggregate(JsonObject, String, String, JsonUnivariateAggregate) getMvAggregate()
     */
    public static <K extends JsonElement> 
    BiFunction<List<JsonObject>, K, JsonObject> mvAggregateList(
        String resultPartitionKeyProperty, String resultProperty,
        List<Pair<String, JsonUnivariateAggregate[]>> aggregateSpecs) {
      
      BiFunction<List<JsonObject>, K, JsonObject> function = 
        (joList, partition) -> {
          JsonObject joResult = new JsonObject();
          joResult.add(resultPartitionKeyProperty, partition);
          
          JsonObject aggregateResults = new JsonObject();
          joResult.add(resultProperty, aggregateResults);
          
          for (Pair<String, JsonUnivariateAggregate[]> p : aggregateSpecs) {
            String variableName = p.getFirst();
            JsonUnivariateAggregate[] aggregates = p.getSecond();
            
            // Compute the aggregates for the variable
            JsonObject jo2 = JsonAnalytics.aggregateList(resultPartitionKeyProperty,
                resultProperty, jo -> jo.get(variableName).getAsDouble(),
                aggregates).apply(joList,  partition);
            
            // Add the variable's aggregates result to the result
            aggregateResults.add(variableName, jo2.get(resultProperty).getAsJsonObject());
          }
        
          return joResult;
        };
        
        return function;
    }
    
    /**
     * Get the value of an aggregate computed by a multi-variable aggregation.
     * <P>
     * This convenience method can be used to extract information from a JSON object
     * created by {@link #mvAggregateList(String, String, List) mvAggregationList()}
     * or {@link #mvAggregate(TWindow, String, String, List) mvAggregate()}
     * </P>
     * <P>
     * Sample use:
     * <pre>{@code
     * ...
     * TStream<JsonObject> aggData = mvAggregate(window, "id", "aggResults", aggSpecs);
     * 
     * // Create a stream of JsonObject tuples with just the average "tx"
     * TStream<JsonObject> avgTx = aggResults.map(
     *    jo -> {
     *      JsonObject result = new JsonObject();
     *      result.add(partitionKeyName, jo.get(partitionKeyName))
     *      result.add("avgTx", getMvAggregate(jo, "aggResults", "tx", Statistic.MEAN);
     *      return result;
     *    });
     * }</pre>
     * 
     * @param jo a JSON object created by {@code mvAggregationList}
     * @param resultProperty the corresponding value passed to {@code mvAggragateList}
     * @param variableName the data variable of interest in the multivariable aggregates
     * @param aggregate the variable's aggregate of interest
     * @return the variable's aggregate's value as a JsonElement
     * @throws RuntimeException if the aggregate isn't present in the result
     * 
     * @see #hasMvAggregate(JsonObject, String, String, JsonUnivariateAggregate) hasAggregate()
     * @see #mvAggregate(TWindow, String, String, List) mvAggregate()
     * @see #mvAggregateList(String, String, List) mvAggregateList()
     */
    public static JsonElement getMvAggregate(JsonObject jo, String resultProperty, String variableName, JsonUnivariateAggregate aggregate) {
      return jo.get(resultProperty).getAsJsonObject()
              .get(variableName).getAsJsonObject()
              .get(aggregate.name());
    }

    /**
     * Check if an aggregation result from a multi-variable aggregation
     * is present.
     * 
     * @param jo a JSON object created by {@code mvAggregationList}
     * @param resultProperty the corresponding value passed to {@code mvAggragateList}
     * @param variableName the data variable of interest in the multivariable aggregates
     * @param aggregate the variable's aggregate of interest
     * @return true if the specified aggregate is present in the jo, false otherwise.
     * 
     * @see #getMvAggregate(JsonObject, String, String, JsonUnivariateAggregate) getMvAggregate()
     */
    public static boolean hasMvAggregate(JsonObject jo, String resultProperty, String variableName, JsonUnivariateAggregate aggregate) {
      JsonElement je = jo.get(resultProperty);
      if (je != null && je.isJsonObject()) {
        JsonObject jo2 = je.getAsJsonObject();
        je = jo2.get(variableName);
        if (je != null && je.isJsonObject()) {
          jo2 = je.getAsJsonObject();
          je = jo2.get(aggregate.name());
          if (je != null)
            return true;
        }
      }
      return false;
    }

}