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.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.Objects;

/**
 * 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 simple KeySelector to divide by key (address) the lamp data
     */
    public static final class EmberLampAddressSelector implements KeySelector<Tuple2<EmberInput.StreetLamp, String>, String> {

        /**
         * @param streetLampByAddress the tuple <{@link it.uniroma2.ember.EmberInput.StreetLamp}, address>
         * @return address as a key
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<EmberInput.StreetLamp, String> streetLampByAddress) throws Exception {
            return streetLampByAddress.f1;
        }
    }

    /**
     * Implements a simple KeySelector to divide by key (address) the aggregated sensors data
     */
    public static final class EmberSensorsAddressSelector implements KeySelector<Tuple2<String,Tuple2<Float,Float>>, String> {

        /**
         * @param aggregatedSensorsData the tuple <address, Tuple2<light_value,traffic_value>>
         * @return address as a key
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Tuple2<Float, Float>> aggregatedSensorsData) throws Exception {
            return aggregatedSensorsData.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>, Tuple2<String, Tuple2<Float,Float>>> {

        @Override
        public Tuple2<String, Tuple2<Float,Float>> join(Tuple2<String, Float> trafficData, Tuple2<String, Float> lumenData) throws Exception {
            if (Objects.equals(lumenData.f0, trafficData.f0)) {
                return new Tuple2<>(lumenData.f0, new Tuple2<>(lumenData.f1, trafficData.f1));
            } else {
                return new Tuple2<>("null", new Tuple2<>(null,null));
            }
        }
    }
}

@@ -1427,613 +1427,8 @@
  %7D%0A%0A
-    /**%0A     * Implements a simple KeySelector to divide by key (address) the lamp data%0A     */%0A    public static final class EmberLampAddressSelector implements KeySelector%3CTuple2%3CEmberInput.StreetLamp, String%3E, String%3E %7B%0A%0A        /**%0A         * @param streetLampByAddress the tuple %3C%7B@link it.uniroma2.ember.EmberInput.StreetLamp%7D, address%3E%0A         * @return address as a key%0A         * @throws Exception%0A         */%0A        @Override%0A        public String getKey(Tuple2%3CEmberInput.StreetLamp, String%3E streetLampByAddress) throws Exception %7B%0A            return streetLampByAddress.f1;%0A        %7D%0A    %7D%0A%0A
     
@@ -1525,16 +1525,16 @@
 rs data%0A
+
      */%0A
@@ -2035,28 +2035,24 @@
    %7D%0A%0A%0A%0A
-    
     /**%0A
      * I
@@ -2035,32 +2035,32 @@
    %7D%0A%0A%0A%0A    /**%0A
+
      * Implement
@@ -2708,24 +2708,24 @@
 ull,null));%0A
-
             
@@ -2741,11 +2741,1106 @@
  %7D%0A    %7D
+%0A%0A    /**%0A     * Implements a join function to set a proper level for the lamp%0A     * (this class actually implements the control feedback)%0A     */%0A    public static final class  EmberControlRoom implements JoinFunction%3CEmberInput.StreetLamp, Tuple2%3CString, Tuple2%3CFloat,Float%3E%3E,EmberInput.StreetLamp%3E %7B%0A%0A        /**%0A         * @param streetLamp a %7B@link it.uniroma2.ember.EmberInput.StreetLamp%7D instance%0A         * @param aggregatedSensorsData a %3Caddress, Tuple2%3Clight_value,traffic_value%3E%3E%0A         * @return %7B@link it.uniroma2.ember.EmberInput.StreetLamp%7D instance with the correct level value%0A         * @throws Exception%0A         */%0A        @Override%0A        public EmberInput.StreetLamp join(EmberInput.StreetLamp streetLamp, Tuple2%3CString, Tuple2%3CFloat, Float%3E%3E aggregatedSensorsData) throws Exception %7B%0A            if (!Objects.equals(aggregatedSensorsData.f0, %22null%22)) %7B%0A                streetLamp.setLevel(aggregatedSensorsData.f1.f0 - aggregatedSensorsData.f1.f1); //TODO%0A                return streetLamp;%0A            %7D%0A            return new EmberInput.StreetLamp();%0A        %7D%0A    %7D
 %0A%7D%0A

Add join class for control room method