package reactiontest.compare;

import java.util.Date;
import java.util.List;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;

import reactiontest.offline.BatchFunctions;
import reactiontest.online.StreamFunctions;

/**
 * Compare the online statistics with the offline computed statistics 
 * (integrate in the online analysis the offline values) 
 *
 */
public class MainCompare {

	public static void main(String[] args) throws Exception {
		
		// Condition for following metrics:
		// 1. running Apache Kafka
		// 2. insert data to Kafka by copy/paste following JSON string:
		// {"medicalid":"Fabio", "operationissue":"italia", "age":33, "gender":"Male", "datetime":"2016-11-03 20:59:28.807", "type":"PreOperation", "times":[141,1750,2000] }
		
		
		// Prediction 1:
		printPredictionByAverage();
		
		// Prediction 2:
		printPredictionByAVGofMedians();
		
		// Prediction 3:
		printPredictionBySlidingAverage();
		
	}

	private static void printPredictionByAVGofMedians() throws Exception {
		// Prediction 2: Median off all reaction data since October 2016 
		// and last reaction times specified by tumbling window
		BatchFunctions batch = new BatchFunctions();
		DataSet<Tuple2<Double, Integer>> dataSet1 = batch.loadDataSetOfOctober2016();
		double median = batch.getMedianReactionTime(dataSet1);
		
		
		StreamFunctions stream = new StreamFunctions();
		DataStream<Tuple7<String, String, Integer, String, Date, String, List<Double>>> dataStream1 = stream.getKafkaStream();
		stream.printPredictionForNextReactionTimeByMedians(dataStream1, median, Time.seconds(10));
		stream.execute();
	}

	private static void printPredictionByAverage() throws Exception {
		// Prediction 1: Average off October 2016 reaction data + tumbling window
		BatchFunctions human = new BatchFunctions();
		DataSet<Tuple2<Double, Integer>> dataSet1 = human.loadDataSetOfOctober2016();
		double average = human.getAverageReaction(dataSet1);
		
		
		StreamFunctions stream = new StreamFunctions();
		//START KAFKA Broker and Zookeeper
		DataStream<Tuple7<String, String, Integer, String, Date, String, List<Double>>> dataStream1 = stream.getKafkaStream();
		stream.printPredictionForNextReactionTimeByAVGs(dataStream1, average, Time.seconds(10));
		stream.execute();
	}
	
	private static void printPredictionBySlidingAverage() throws Exception {
		// Prediction 1: Average off October 2016 reaction data + tumbling window
		BatchFunctions human = new BatchFunctions();
		DataSet<Tuple2<Double, Integer>> dataSet1 = human.loadDataSetOfOctober2016();
		double average = human.getAverageReaction(dataSet1);
		
		
		StreamFunctions stream = new StreamFunctions();
		DataStream<Tuple7<String, String, Integer, String, Date, String, List<Double>>> dataStream1 = stream.getKafkaStream();
		stream.printPredictionForNextReactionTimeBySlidingAVGs(dataStream1, average, Time.seconds(10), Time.seconds(3));
		stream.execute();
	}

}

@@ -1363,32 +1363,91 @@
 tchFunctions();%0A
+%09%09%0A%09%09// first get median of reaction times by offline data%0A
 %09%09DataSet%3CTuple2
@@ -1564,32 +1564,84 @@
 dataSet1);%0A%09%09%0A%09%09
+// second use calculated median to make a prediction
 %0A%09%09StreamFunctio
@@ -1788,32 +1788,145 @@
 tKafkaStream();%0A
+%09%09%0A%09%09// prediction for next reaction time = Median of the offlineAverage and onlineAverage using tumbling window%0A
 %09%09stream.printPr
@@ -2161,32 +2161,88 @@
 tumbling window%0A
+%09%09%0A%09%09// first get average reaction time by offline data%0A
 %09%09BatchFunctions
@@ -2234,37 +2234,37 @@
 %09BatchFunctions 
-human
+batch
  = new BatchFunc
@@ -2263,32 +2263,33 @@
 atchFunctions();
+ 
 %0A%09%09DataSet%3CTuple
@@ -2311,37 +2311,37 @@
 er%3E%3E dataSet1 = 
-human
+batch
 .loadDataSetOfOc
@@ -2364,37 +2364,37 @@
 ouble average = 
-human
+batch
 .getAverageReact
@@ -2405,32 +2405,85 @@
 dataSet1);%0A%09%09%0A%09%09
+// second use calculated average to make a prediction
 %0A%09%09StreamFunctio
@@ -2521,45 +2521,8 @@
 ();%0A
-%09%09//START KAFKA Broker and Zookeeper%0A
 %09%09Da
@@ -2630,32 +2630,143 @@
 tKafkaStream();%0A
+%09%09%0A%09%09// prediction for next reaction time = Mean of the offlineAverage and onlineAverage using tumbling window%0A
 %09%09stream.printPr
@@ -2947,33 +2947,33 @@
 %09%09// Prediction 
-1
+3
 : Average off Oc
@@ -3007,32 +3007,88 @@
 tumbling window%0A
+%09%09%0A%09%09// first get average reaction time by offline data%0A
 %09%09BatchFunctions
@@ -3258,16 +3258,69 @@
 );%0A%09%09%0A%09%09
+// second use calculated average to make a prediction
 %0A%09%09Strea
@@ -3483,16 +3483,123 @@
 ream();%0A
+%09%09// prediction for next reaction time = Mean of the offlineAverage and onlineAverage using sliding window%0A
 %09%09stream

fix #15 add some comments