D-Net

package eu.dnetlib.iis.referenceextraction.project;

import java.util.HashMap;

import java.util.Map;

import org.apache.avro.file.DataFileWriter;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.log4j.Logger;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.ProcessUtils;

import eu.dnetlib.iis.core.java.io.CloseableIterator;

import eu.dnetlib.iis.core.java.io.DataStore;

import eu.dnetlib.iis.core.java.io.FileSystemPath;

import eu.dnetlib.iis.core.java.porttype.AvroPortType;

import eu.dnetlib.iis.core.java.porttype.PortType;

import eu.dnetlib.iis.referenceextraction.project.schemas.DocumentToProject;

/**

 * Confidence level based {@link DocumentToProject} relations cleaner.

 * @author mhorst

 *

 */

public class DocumentToProjectConfidenceLevelBasedCleaner implements eu.dnetlib.iis.core.java.Process {

	public static final String CONFIDENCE_LEVEL_THRESHOLD = "export.document_to_project.confidence.level.threshold";

	private final Logger log = Logger.getLogger(this.getClass());

	private final static String inputPort = "input";

	private final static String outputPort = "output";

	@Override

	public Map<String, PortType> getInputPorts() {

		return createInputPorts();

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return createOutputPorts();

	}

	private static HashMap<String, PortType> createInputPorts(){

		HashMap<String, PortType> inputPorts = 

				new HashMap<String, PortType>();

		inputPorts.put(inputPort, 

				new AvroPortType(DocumentToProject.SCHEMA$));

		return inputPorts;

	}

	private static HashMap<String, PortType> createOutputPorts(){

		HashMap<String, PortType> outputPorts = 

				new HashMap<String, PortType>();

		outputPorts.put(outputPort, 

				new AvroPortType(DocumentToProject.SCHEMA$));

		return outputPorts;	

	}

	@Override

	public void run(PortBindings portBindings, Configuration conf,

			Map<String, String> parameters) throws Exception {

		String confidenceLevelThresholdStr = ProcessUtils.getParameterValue(

				CONFIDENCE_LEVEL_THRESHOLD, 

				conf, parameters);

		if (confidenceLevelThresholdStr==null || confidenceLevelThresholdStr.isEmpty()) {

			throw new RuntimeException("no confidence level threshold "

					+ "parameter provided: '" + CONFIDENCE_LEVEL_THRESHOLD + "'");

		}

		float confidenceLevelThreshold = Float.parseFloat(confidenceLevelThresholdStr);

		Map<String, Path> input = portBindings.getInput();

		Map<String, Path> output = portBindings.getOutput();

		FileSystem fs = FileSystem.get(conf);

		CloseableIterator<DocumentToProject> it = DataStore.getReader(

				new FileSystemPath(fs, input.get(inputPort)));

		DataFileWriter<DocumentToProject> writer = DataStore.create(

				new FileSystemPath(fs, output.get(outputPort)), 

				DocumentToProject.SCHEMA$);

		try {

			while (it.hasNext()) {

				DocumentToProject current = it.next();

				if (current.getConfidenceLevel()==null ||

						current.getConfidenceLevel()>=confidenceLevelThreshold) {

					writer.append(current);

				} else {

					log.warn("skipping relation, "

							+ "confidence level below the threshold "

							+ "("+ confidenceLevelThresholdStr	+"): " + 

							current.toString());

				}

			}	

		} finally {

			it.close();

			writer.close();

		}

	}

}