started to generalize averaging aggregators; implemented for Distance

Change-Id: I54c7c03810626246f12cfe6bca15a659e066add0

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/AbstractParallelAverageAggregationProcessor.java

@@ -0,0 +1,80 @@
+package com.sap.sailing.datamining.impl.components.aggregators;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutorService;
+
+import com.sap.sse.datamining.components.AggregationProcessorDefinition;
+import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.components.GroupedDataEntry;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelComparableMaxAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelComparableMinAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.shared.GroupKey;
+import com.sap.sse.datamining.shared.data.AverageWithStats;
+import com.sap.sse.datamining.shared.impl.AverageWithStatsImpl;
+
+public abstract class AbstractParallelAverageAggregationProcessor<T extends Comparable<T>>
+            extends AbstractParallelGroupedDataStoringAggregationProcessor<T, AverageWithStats<T>> {
+
+    private final AbstractParallelSumAggregationProcessor<T> sumAggregationProcessor;
+    private final AbstractParallelComparableMinAggregationProcessor<T> minAggregationProcessor;
+    private final AbstractParallelComparableMaxAggregationProcessor<T> maxAggregationProcessor;
+    private final Map<GroupKey, Integer> elementAmountPerKey;
+
+    public AbstractParallelAverageAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, AverageWithStats<T>>, ?>> resultReceivers,
+            AbstractParallelSumAggregationProcessor<T> sumAggregationProcessor,
+            AbstractParallelComparableMinAggregationProcessor<T> minAggregationProcessor,
+            AbstractParallelComparableMaxAggregationProcessor<T> maxAggregationProcessor) {
+        super(executor, resultReceivers, "Average");
+        elementAmountPerKey = new HashMap<>();
+        this.sumAggregationProcessor = sumAggregationProcessor;
+        this.minAggregationProcessor = minAggregationProcessor;
+        this.maxAggregationProcessor = maxAggregationProcessor;
+    }
+
+    public abstract AggregationProcessorDefinition<T, AverageWithStats<T>> getProcessorDefinition();
+
+    @Override
+    protected void storeElement(GroupedDataEntry<T> element) {
+        incrementElementAmount(element);
+        sumAggregationProcessor.handleElement(element);
+        minAggregationProcessor.handleElement(element);
+        maxAggregationProcessor.handleElement(element);
+    }
+
+    private void incrementElementAmount(GroupedDataEntry<T> element) {
+        GroupKey key = element.getKey();
+        if (!elementAmountPerKey.containsKey(key)) {
+            elementAmountPerKey.put(key, 0);
+        }
+        Integer currentAmount = elementAmountPerKey.get(key);
+        elementAmountPerKey.put(key, currentAmount + 1);
+    }
+    
+    protected abstract T divide(T sum, long elementCount);
+
+    @Override
+    protected Map<GroupKey, AverageWithStats<T>> aggregateResult() {
+        Map<GroupKey,  AverageWithStats<T>> result = new HashMap<>();
+        Map<GroupKey, T> sumAggregation = sumAggregationProcessor.getResult();
+        Map<GroupKey, T> minAggregation = minAggregationProcessor.getResult();
+        Map<GroupKey, T> maxAggregation = maxAggregationProcessor.getResult();
+        for (Entry<GroupKey, T> sumAggregationEntry : sumAggregation.entrySet()) {
+            GroupKey key = sumAggregationEntry.getKey();
+            result.put(key, new AverageWithStatsImpl<>(
+                    /* average */ divide(sumAggregationEntry.getValue(), elementAmountPerKey.get(key).longValue()),
+                    /* min */ minAggregation.get(key),
+                    /* max */ maxAggregation.get(key),
+                    /* median */ null,
+                    /* standardDeviation */ null,
+                    /* count */ elementAmountPerKey.get(key).longValue(),
+                    getProcessorDefinition().getExtractedType().getName()));
+        }
+        return result;
+    }
+
+}

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/AbstractParallelSumAggregationProcessor.java

@@ -0,0 +1,39 @@
+package com.sap.sailing.datamining.impl.components.aggregators;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.components.GroupedDataEntry;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
+import com.sap.sse.datamining.shared.GroupKey;
+
+public abstract class AbstractParallelSumAggregationProcessor<T> extends
+        AbstractParallelGroupedDataAggregationProcessor<T, T> {
+    
+    private final Map<GroupKey, T> results;
+
+    public AbstractParallelSumAggregationProcessor(ExecutorService executor, Collection<Processor<Map<GroupKey, T>, ?>> resultReceivers) {
+        super(executor, resultReceivers, "Sum");
+        results = new HashMap<>();
+    }
+    
+    protected abstract T add(T t1, T t2);
+
+    @Override
+    protected void handleElement(GroupedDataEntry<T> element) {
+        GroupKey key = element.getKey();
+        if (!results.containsKey(key)) {
+            results.put(key, element.getDataEntry());
+        } else {
+            results.put(key, add(results.get(key), element.getDataEntry()));
+        }
+    }
+
+    @Override
+    protected Map<GroupKey, T> getResult() {
+        return results;
+    }
+}

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelDistanceAverageAggregationProcessor.java

@@ -1,63 +1,44 @@
 package com.sap.sailing.datamining.impl.components.aggregators;
 
 import java.util.Collection;
-import java.util.HashMap;
+import java.util.Collections;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.concurrent.ExecutorService;
 
 import com.sap.sailing.domain.common.Distance;
 import com.sap.sse.datamining.components.AggregationProcessorDefinition;
 import com.sap.sse.datamining.components.Processor;
-import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
+import com.sap.sse.datamining.shared.data.AverageWithStats;
 
 public class ParallelDistanceAverageAggregationProcessor
-            extends AbstractParallelGroupedDataStoringAggregationProcessor<Distance, Distance> {
+            extends AbstractParallelAverageAggregationProcessor<Distance> {
+    private static final Class<?> _c = AverageWithStats.class;
+    @SuppressWarnings("unchecked")
+    private static final Class<AverageWithStats<Distance>> _cc = (Class<AverageWithStats<Distance>>) _c;
+
+    private static final AggregationProcessorDefinition<Distance, AverageWithStats<Distance>> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Distance.class, _cc, "Average", ParallelDistanceAverageAggregationProcessor.class);
     
-    private static final AggregationProcessorDefinition<Distance, Distance> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Distance.class, Distance.class, "Average", ParallelDistanceAverageAggregationProcessor.class);
-    
-    public static AggregationProcessorDefinition<Distance, Distance> getDefinition() {
+    public ParallelDistanceAverageAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, AverageWithStats<Distance>>, ?>> resultReceivers) {
+        super(executor, resultReceivers, new ParallelDistanceSumAggregationProcessor(executor, /* resultReceivers */ Collections.emptyList()),
+                new ParallelDistanceMinAggregationProcessor(executor, /* resultReceivers */ Collections.emptyList()),
+                new ParallelDistanceMaxAggregationProcessor(executor, /* resultReceivers */ Collections.emptyList()));
+    }
+
+    public static AggregationProcessorDefinition<Distance, AverageWithStats<Distance>> getDefinition() {
         return DEFINITION;
     }
 
-    private final ParallelDistanceSumAggregationProcessor sumAggregationProcessor;
-    private final Map<GroupKey, Integer> elementAmountPerKey;
-
-    public ParallelDistanceAverageAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, Distance>, ?>> resultReceivers) {
-        super(executor, resultReceivers, "Average");
-        elementAmountPerKey = new HashMap<>();
-        sumAggregationProcessor = new ParallelDistanceSumAggregationProcessor(executor, resultReceivers);
+    @Override
+    public AggregationProcessorDefinition<Distance, AverageWithStats<Distance>> getProcessorDefinition() {
+        return DEFINITION;
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Distance> element) {
-        incrementElementAmount(element);
-        sumAggregationProcessor.handleElement(element);
+    protected Distance divide(Distance sum, long elementCount) {
+        return sum.scale(1. / (double) elementCount);
     }
-
-    private void incrementElementAmount(GroupedDataEntry<Distance> element) {
-        GroupKey key = element.getKey();
-        if (!elementAmountPerKey.containsKey(key)) {
-            elementAmountPerKey.put(key, 0);
-        }
-        Integer currentAmount = elementAmountPerKey.get(key);
-        elementAmountPerKey.put(key, currentAmount + 1);
-    }
-
-    @Override
-    protected Map<GroupKey, Distance> aggregateResult() {
-        Map<GroupKey, Distance> result = new HashMap<>();
-        Map<GroupKey, Distance> sumAggregation = sumAggregationProcessor.getResult();
-        for (Entry<GroupKey, Distance> sumAggregationEntry : sumAggregation.entrySet()) {
-            GroupKey key = sumAggregationEntry.getKey();
-            result.put(key, sumAggregationEntry.getValue().scale(1. / elementAmountPerKey.get(key).doubleValue()));
-        }
-        return result;
-    }
-
 }

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelDistanceSumAggregationProcessor.java

@@ -1,20 +1,17 @@
 package com.sap.sailing.datamining.impl.components.aggregators;
 
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
 import com.sap.sailing.domain.common.Distance;
 import com.sap.sse.datamining.components.AggregationProcessorDefinition;
 import com.sap.sse.datamining.components.Processor;
-import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 
 public class ParallelDistanceSumAggregationProcessor extends
-        AbstractParallelGroupedDataAggregationProcessor<Distance, Distance> {
+        AbstractParallelSumAggregationProcessor<Distance> {
     
     private static final AggregationProcessorDefinition<Distance, Distance> DEFINITION =
             new SimpleAggregationProcessorDefinition<>(Distance.class, Distance.class, "Sum", ParallelDistanceSumAggregationProcessor.class);
@@ -23,25 +20,12 @@ public class ParallelDistanceSumAggregationProcessor extends
         return DEFINITION;
     }
     
-    private final Map<GroupKey, Distance> results;
-
     public ParallelDistanceSumAggregationProcessor(ExecutorService executor, Collection<Processor<Map<GroupKey, Distance>, ?>> resultReceivers) {
-        super(executor, resultReceivers, "Sum");
-        results = new HashMap<>();
+        super(executor, resultReceivers);
     }
 
     @Override
-    protected void handleElement(GroupedDataEntry<Distance> element) {
-        GroupKey key = element.getKey();
-        if (!results.containsKey(key)) {
-            results.put(key, element.getDataEntry());
-        } else {
-            results.put(key, results.get(key).add(element.getDataEntry()));
-        }
-    }
-
-    @Override
-    protected Map<GroupKey, Distance> getResult() {
-        return results;
+    protected Distance add(Distance t1, Distance t2) {
+        return t1.add(t2);
     }
 }

java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/datamining/presentation/ResultsChart.java

@@ -275,7 +275,7 @@ public class ResultsChart extends AbstractNumericResultsPresenter<Settings> {
             point.setName(mainKey.asString());
             seriesMappedByGroupKey.get(groupKeyToSeriesKey(resultEntry.getKey()))
                 .addPoint(point, false, false, false);
-            final Triple<Number, Number, Long> errorMargins = currentResultErrorMargins.get(mainKey);
+            final Triple<Number, Number, Long> errorMargins = currentResultErrorMargins==null?null:currentResultErrorMargins.get(mainKey);
             if (errorMargins != null) {
                 Point errorMarginsPoint = new Point(mainKeyToXValueMap.get(mainKey), errorMargins.getA(), errorMargins.getB());
                 errorMarginsPoint.setName(mainKey.asString()+", "+stringMessages.elements(errorMargins.getC()));

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelSingleGroupedValueAggregationProcessor.java

@@ -27,7 +27,7 @@ public abstract class AbstractParallelSingleGroupedValueAggregationProcessor<Val
     }
 
     @Override
-    protected void handleElement(GroupedDataEntry<ValueType> element) {
+    public void handleElement(GroupedDataEntry<ValueType> element) {
         GroupKey key = element.getKey();
         ValueType value = element.getDataEntry();
         if (value != null) {
@@ -47,7 +47,7 @@ public abstract class AbstractParallelSingleGroupedValueAggregationProcessor<Val
     protected abstract ValueType compareValuesAndReturnNewResult(ValueType currentResult, ValueType newValue);
 
     @Override
-    protected Map<GroupKey, ValueType> getResult() {
+    public Map<GroupKey, ValueType> getResult() {
         return valueMap;
     }