// Path: disambiguation-author/disambiguation-author-logic/src/main/java/pl/edu/icm/coansys/disambiguation/model/ContributorWithExtractedFeatures.java
// public class ContributorWithExtractedFeatures implements Serializable{
//     String docKey;
//     String contributorId;
//     Map<String,Collection<Integer>> metadata;
//     Integer surnameInt;
//     String surnameString;
//     boolean surnameNotNull;
// 
//     public ContributorWithExtractedFeatures(String docKey, String contributorId, Map<String, Collection<Integer>> metadata, Integer surnameInt, String surnameString, boolean surnameNotNull) {
//         this.docKey = docKey;
//         this.contributorId = contributorId;
//         this.metadata = metadata;
//         this.surnameInt = surnameInt;
//         this.surnameString = surnameString;
//         this.surnameNotNull = surnameNotNull;
//     }
// 
//     public String getDocKey() {
//         return docKey;
//     }
// 
//     public String getContributorId() {
//         return contributorId;
//     }
// 
//     public Map<String, Collection<Integer>> getMetadata() {
//         return metadata;
//     }
// 
//     public Integer getSurnameInt() {
//         return surnameInt;
//     }
// 
//     public String getSurnameString() {
//         return surnameString;
//     }
// 
//     public boolean isSurnameNotNull() {
//         return surnameNotNull;
//     }
//     
//     public Tuple asTuple(){
//         Object[] to = new Object[] { docKey, contributorId, surnameInt,
// 						hashMapToDataBag(), surnameString, isSurnameNotNull() };
// 		return  TupleFactory.getInstance()
// 						.newTuple(Arrays.asList(to));
//     }
//     
//     private HashMap<Object,DataBag> hashMapToDataBag(){
//         TupleFactory tf=TupleFactory.getInstance();
//         HashMap<Object,DataBag> ret=new HashMap<>();
//         for (Map.Entry<String,Collection<Integer>> entry:metadata.entrySet()){
//            DataBag db = new DefaultDataBag();
//            for (Integer i:entry.getValue()) {
//                db.add(tf.newTuple((Object) i));
//            }
//            ret.put(entry.getKey(), db);
//         }
//         return ret;
//         
//     }
//     
//     
// }


import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.Tuple;
import org.apache.pig.tools.pigstats.PigStatusReporter;
import org.slf4j.LoggerFactory;
import pl.edu.icm.coansys.commons.java.StackTraceExtractor;
import pl.edu.icm.coansys.disambiguation.model.ContributorWithExtractedFeatures;
/*
 * This file is part of CoAnSys project.
 * Copyright (c) 2012-2015 ICM-UW
 * 
 * CoAnSys is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.

 * CoAnSys is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU Affero General Public License for more details.
 * 
 * You should have received a copy of the GNU Affero General Public License
 * along with CoAnSys. If not, see <http://www.gnu.org/licenses/>.
 */

package pl.edu.icm.coansys.disambiguation.author.pig;




/**
 * Verify that the author may be similar (comparable) to someone by checking
 * whether there is a minimum number of features.
 * 
 * @author mwos
 */
@SuppressWarnings({ "unchecked" })
public class FeaturesCheck extends AND<Boolean> {

	private static final org.slf4j.Logger logger = LoggerFactory
			.getLogger(FeaturesCheck.class);
	// benchmark
	private boolean isStatistics = false;
	private static int skipedContribCounter = 0, allContribCounter = 0;

	/**
	 * @param Tuple
	 *            input with cid, sname, map with features
	 * @throws IllegalAccessException
	 * @throws InstantiationException
	 * @throws ClassNotFoundException
	 * @throws Exception
	 * @throws ExecException
	 * @returns String UUID
	 */

	public FeaturesCheck(String threshold, String featureDescription,
			String useIdsForExtractors, String printStatistics)
			throws ClassNotFoundException, InstantiationException,
			IllegalAccessException {
		super(logger, threshold, featureDescription, useIdsForExtractors);
		this.isStatistics = Boolean.parseBoolean(printStatistics);
	}
     static ConcurrentHashMap<String,FeaturesCheck> extractors=new ConcurrentHashMap<>();
    
    public static synchronized FeaturesCheck getFeaturesCheck(String  threshold, String featureDescription,
			String useIdsForExtractors, String printStatistics) throws InstantiationException, IllegalAccessException, ClassNotFoundException {
        String key=threshold+featureDescription+useIdsForExtractors+printStatistics;
        if (!extractors.containsKey(key)) {
            extractors.put(key, new FeaturesCheck(threshold, featureDescription, useIdsForExtractors, printStatistics));
        }
        return extractors.get(key);
    }
    
    
// Path: disambiguation-author/disambiguation-author-logic/src/main/java/pl/edu/icm/coansys/disambiguation/model/ContributorWithExtractedFeatures.java
// public class ContributorWithExtractedFeatures implements Serializable{
//     String docKey;
//     String contributorId;
//     Map<String,Collection<Integer>> metadata;
//     Integer surnameInt;
//     String surnameString;
//     boolean surnameNotNull;
// 
//     public ContributorWithExtractedFeatures(String docKey, String contributorId, Map<String, Collection<Integer>> metadata, Integer surnameInt, String surnameString, boolean surnameNotNull) {
//         this.docKey = docKey;
//         this.contributorId = contributorId;
//         this.metadata = metadata;
//         this.surnameInt = surnameInt;
//         this.surnameString = surnameString;
//         this.surnameNotNull = surnameNotNull;
//     }
// 
//     public String getDocKey() {
//         return docKey;
//     }
// 
//     public String getContributorId() {
//         return contributorId;
//     }
// 
//     public Map<String, Collection<Integer>> getMetadata() {
//         return metadata;
//     }
// 
//     public Integer getSurnameInt() {
//         return surnameInt;
//     }
// 
//     public String getSurnameString() {
//         return surnameString;
//     }
// 
//     public boolean isSurnameNotNull() {
//         return surnameNotNull;
//     }
//     
//     public Tuple asTuple(){
//         Object[] to = new Object[] { docKey, contributorId, surnameInt,
// 						hashMapToDataBag(), surnameString, isSurnameNotNull() };
// 		return  TupleFactory.getInstance()
// 						.newTuple(Arrays.asList(to));
//     }
//     
//     private HashMap<Object,DataBag> hashMapToDataBag(){
//         TupleFactory tf=TupleFactory.getInstance();
//         HashMap<Object,DataBag> ret=new HashMap<>();
//         for (Map.Entry<String,Collection<Integer>> entry:metadata.entrySet()){
//            DataBag db = new DefaultDataBag();
//            for (Integer i:entry.getValue()) {
//                db.add(tf.newTuple((Object) i));
//            }
//            ret.put(entry.getKey(), db);
//         }
//         return ret;
//         
//     }
//     
//     
// }


// Path: disambiguation-author/disambiguation-author-logic/src/main/java/pl/edu/icm/coansys/disambiguation/author/pig/FeaturesCheck.java
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.Tuple;
import org.apache.pig.tools.pigstats.PigStatusReporter;
import org.slf4j.LoggerFactory;
import pl.edu.icm.coansys.commons.java.StackTraceExtractor;
import pl.edu.icm.coansys.disambiguation.model.ContributorWithExtractedFeatures;

/*
 * This file is part of CoAnSys project.
 * Copyright (c) 2012-2015 ICM-UW
 * 
 * CoAnSys is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.

 * CoAnSys is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU Affero General Public License for more details.
 * 
 * You should have received a copy of the GNU Affero General Public License
 * along with CoAnSys. If not, see <http://www.gnu.org/licenses/>.
 */

package pl.edu.icm.coansys.disambiguation.author.pig;




/**
 * Verify that the author may be similar (comparable) to someone by checking
 * whether there is a minimum number of features.
 * 
 * @author mwos
 */
@SuppressWarnings({ "unchecked" })
public class FeaturesCheck extends AND<Boolean> {

	private static final org.slf4j.Logger logger = LoggerFactory
			.getLogger(FeaturesCheck.class);
	// benchmark
	private boolean isStatistics = false;
	private static int skipedContribCounter = 0, allContribCounter = 0;

	/**
	 * @param Tuple
	 *            input with cid, sname, map with features
	 * @throws IllegalAccessException
	 * @throws InstantiationException
	 * @throws ClassNotFoundException
	 * @throws Exception
	 * @throws ExecException
	 * @returns String UUID
	 */

	public FeaturesCheck(String threshold, String featureDescription,
			String useIdsForExtractors, String printStatistics)
			throws ClassNotFoundException, InstantiationException,
			IllegalAccessException {
		super(logger, threshold, featureDescription, useIdsForExtractors);
		this.isStatistics = Boolean.parseBoolean(printStatistics);
	}
     static ConcurrentHashMap<String,FeaturesCheck> extractors=new ConcurrentHashMap<>();
    
    public static synchronized FeaturesCheck getFeaturesCheck(String  threshold, String featureDescription,
			String useIdsForExtractors, String printStatistics) throws InstantiationException, IllegalAccessException, ClassNotFoundException {
        String key=threshold+featureDescription+useIdsForExtractors+printStatistics;
        if (!extractors.containsKey(key)) {
            extractors.put(key, new FeaturesCheck(threshold, featureDescription, useIdsForExtractors, printStatistics));
        }
        return extractors.get(key);
    }
    
    
    public Boolean exec(ContributorWithExtractedFeatures cont){