package edu.jhu.thrax.hadoop.features.annotation;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Reducer;

import edu.jhu.thrax.hadoop.datatypes.Annotation;
import edu.jhu.thrax.hadoop.datatypes.FeaturePair;
import edu.jhu.thrax.hadoop.datatypes.RuleWritable;
import edu.jhu.thrax.util.BackwardsCompatibility;
import edu.jhu.thrax.util.Vocabulary;

public class AnnotationReducer extends Reducer<RuleWritable, Annotation, RuleWritable, FeaturePair> {

  private List<AnnotationFeature> annotationFeatures;

  public AnnotationReducer() {}

  protected void setup(Context context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();
    String vocabulary_path = conf.getRaw("thrax.work-dir") + "vocabulary/part-*";
    Vocabulary.initialize(conf, vocabulary_path);

    String features = BackwardsCompatibility.equivalent(conf.get("thrax.features", ""));

    // Paraphrasing only needs the annotation to be passed through.
    String type = conf.get("thrax.type", "translation");
    if ("paraphrasing".equals(type)) {
      annotationFeatures = new ArrayList<AnnotationFeature>();
      annotationFeatures.add(new AnnotationPassthroughFeature());
    } else {
      annotationFeatures = AnnotationFeatureFactory.getAll(features);
    }

    for (AnnotationFeature af : annotationFeatures)
      af.init(context);
  }

  protected void reduce(RuleWritable key, Iterable<Annotation> values, Context context)
      throws IOException, InterruptedException {
    for (Annotation annotation : values) {
      for (AnnotationFeature f : annotationFeatures) {
        context.write(key, new FeaturePair(Vocabulary.id(f.getName()), f.score(key, annotation)));
      }
    }
  }
}
