package de.tuberlin.dima.schubotz.fse;

import com.google.common.collect.HashMultiset;
import eu.stratosphere.api.java.functions.FlatMapFunction;
import eu.stratosphere.util.Collector;

import java.util.Arrays;
import java.util.Collection;

public class QuerySectionMatcher extends FlatMapFunction<SectionTuple,ResultTuple> {
	/**
	 * Split for tex and keywords
	 */
	final String SPLIT;
	final HashMultiset<String> latexDocsMultiset;
	final HashMultiset<String> keywordDocsMultiset;

	public QuerySectionMatcher (String split, HashMultiset<String> latexDocsMultiset, HashMultiset<String> keywordDocsMultiset) {
		SPLIT = split;
		this.latexDocsMultiset = latexDocsMultiset;
		this.keywordDocsMultiset = keywordDocsMultiset;
	}

	/**
	 * The core method of the MapFunction. Takes an element from the input data set and transforms
	 * it into another element.
	 *
	 * @param value The input value.
	 * @return The value produced by the map function from the input value. 
	 * @throws Exception This method may throw exceptions. Throwing an exception will cause the operation
	 *                   to fail and may trigger recovery.
	 */
	@Override
	public void flatMap(SectionTuple in,Collector<ResultTuple> out) {
		HashMultiset<String> queryLatex;
		HashMultiset<String> queryKeywords;
		double latexScore = 0;
		double keywordScore = 0;
		
		//Construct set of term frequencies for latex and keywords
		HashMultiset<String> sectionLatex = HashMultiset.create(Arrays.asList(in.getLatex().split(SPLIT)));
		HashMultiset<String> sectionKeywords = HashMultiset.create(Arrays.asList(in.getKeywords().split(SPLIT)));
		
		//Loop through queries and calculate tfidf scores
		Collection<QueryTuple> queries = getRuntimeContext().getBroadcastVariable("Queries");
		for (QueryTuple query : queries) {
//boolean debug = false; 
//if (in.getID().contains("5478_1_6") && query.getID().contains("Math-1")) { //DEBUG output
//System.out.println(query.toString());
//System.out.println(in.toString());
//System.out.println(Arrays.asList(in.getLatex().split(SPLIT)));
//debug = true;
//}
			if (!sectionLatex.isEmpty()) {
				queryLatex = HashMultiset.create(Arrays.asList(query.getLatex().split(SPLIT)));
				latexScore = calculateTFIDFScore(queryLatex, sectionLatex, latexDocsMultiset);
			}
			
			if (!sectionKeywords.isEmpty()) {
				queryKeywords = HashMultiset.create(Arrays.asList(query.getKeywords().split(SPLIT)));
				keywordScore = calculateTFIDFScore(queryKeywords, sectionKeywords, keywordDocsMultiset);
			}
			
			
			out.collect(new ResultTuple(query.getID(),in.getID(),(keywordScore/6.36) + latexScore));
		}
	}
	
	/**
	 * @param queryTokens
	 * @param sectionTokens
	 * @param map - map to use: either keywordDocsMap or latexDocsMap
	 * @return
	 */
	private double calculateTFIDFScore(HashMultiset<String> queryTokens, HashMultiset<String> sectionTokens, HashMultiset<String> map) {
		double termTotal = sectionTokens.size(); //total number of terms in current section
		double termFreqDoc; //frequency in current section
		double termFreqTotal; //number of documents that contain the term
		double docTotal = MainProgram.numDocs; 
		
		//Calculations based on http://tfidf.com/
		double tf = 0d; //term frequency
		double idf = 0d; //inverse document frequency
		double total = 0d;
				
		for (String element : queryTokens.elementSet()) {
			termFreqDoc = sectionTokens.count(element);
			termFreqTotal = map.count(element);
			tf = termFreqDoc / termTotal; //can be zero but not undefined
			idf = Math.log(docTotal / (1d + termFreqTotal)); //will never be undefined due to +1
			total += tf * idf;
//if (debug) {
////DEBUG output
//System.out.println("Term: " + element);
//System.out.println("Freq in Doc: " + termFreqDoc);
//System.out.println("Num doc with term: " + termFreqTotal);
//System.out.println("tf: " + tf);
//System.out.println("idf: " + idf);
//System.out.println("total: " + total);
//}
		}
//if (debug) {
//	System.out.println("end total: " + total);
//	System.out.println("END END END END");
//}
		return total;
		
	}
}

@@ -477,16 +477,89 @@
 ltiset;%0A
+%09double latexScore = 0;%0A%09double keywordScore = 0;%0A%09double finalScore =0.;
 %0A%09public
@@ -940,21 +940,18 @@
  @param 
-value
+in
  The inp
@@ -1353,59 +1353,8 @@
 ds;%0A
-%09%09double latexScore = 0;%0A%09%09double keywordScore = 0;
 %0A%09%09%0A
@@ -2277,32 +2277,65 @@
 xDocsMultiset);%0A
+%09%09%09%7D else %7B%0A%09%09%09%09latexScore = 0.;%0A
 %09%09%09%7D%0A%09%09%09%0A%09%09%09if (
@@ -2551,24 +2551,155 @@
 t);%0A%09%09%09%7D
-%0A%09%09%09%0A%09%09%09
+ else %7B%0A%09%09%09%09keywordScore = 0.;%0A%09%09%09%7D%0A%09%09%09finalScore = (keywordScore/6.36) + latexScore;%0A%09%09%09if( finalScore == Double.NaN )%0A%09%09%09%09finalScore = 0;
 %0A%09%09%09out.
@@ -2751,35 +2751,14 @@
 D(),
-(keywordScore/6.36) + latex
+ final
 Scor

Make scoring function fault tolerant