package it.uniroma2.ember;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * This is a utility class for aggregate statistics in Project Ember
 */

public class EmberSensorsAggregation {

    /**
     * Implements a simple KeySelector to divide by key (address) the traffic data - window join
     */
    public static final class EmberTrafficMeanSelector implements KeySelector<Tuple2<String, Float>, String> {

        /**
         * @param trafficData the tuple <key, traffic_intensity>
         * @return address as a key
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Float> trafficData) throws Exception {
            return trafficData.f0;
        }
    }

    /**
     * Implements a simple KeySelector to divide by key (address) the lumen data - window join
     */
    public static final class EmberLumenMeanSelector implements KeySelector<Tuple2<String, Float>, String> {

        /**
         * @param lumenData the tuple <key, light_level>
         * @return address as a key
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Float> lumenData) throws Exception {
            return lumenData.f0;
        }
    }

    /**
     * Implements a join function to aggregate mean from traffic and lumen data streams
     */
    public static final class EmberAggregateSensors implements JoinFunction<Tuple2<String, Float>, Tuple2<String, Float>, Float> {

        @Override
        public Float join(Tuple2<String, Float> trafficData, Tuple2<String, Float> lumenData) throws Exception {
            return Math.abs(lumenData.f1 - trafficData.f1);
        }
    }
}

@@ -1575,23 +1575,39 @@
  Float%3E,
+ Tuple2%3CString,
  Float%3E
+%3E
  %7B%0A%0A    
@@ -1634,22 +1634,38 @@
   public
+ Tuple2%3CString,
  Float
+%3E
  join(Tu
@@ -1767,16 +1767,43 @@
   return
+ new Tuple2%3C%3E(lumenData.f0,
  Math.ab
@@ -1834,16 +1834,17 @@
 Data.f1)
+)
 ;%0A      

Fix a wrong implementation of sensors aggregation.