/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package at.illecker.storm.commons.bolt;

import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import at.illecker.storm.commons.postagger.ArkPOSTagger;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import cmu.arktweetnlp.Tagger.TaggedToken;

public class POSTaggerBolt extends BaseBasicBolt {
  public static final String ID = "pos-tagger-bolt";
  public static final String CONF_LOGGING = ID + ".logging";
  public static final String CONF_MODEL = ID + ".model";
  private static final long serialVersionUID = 8389930087364663504L;
  private static final Logger LOG = LoggerFactory
      .getLogger(POSTaggerBolt.class);
  private boolean m_logging = false;

  private ArkPOSTagger m_tagger;

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // key of output tuples
    declarer.declare(new Fields("taggedTokens"));
  }

  @Override
  public void prepare(Map config, TopologyContext context) {
    // Optional set logging
    if (config.get(CONF_LOGGING) != null) {
      m_logging = (Boolean) config.get(CONF_LOGGING);
    } else {
      m_logging = false;
    }
    // Load ARK POS Tagger
    m_tagger = ArkPOSTagger.getInstance();
  }

  @Override
  public void execute(Tuple tuple, BasicOutputCollector collector) {
    List<String> preprocessedTokens = (List<String>) tuple
        .getValueByField("preprocessedTokens");

    // POS Tagging
    List<TaggedToken> taggedTokens = m_tagger.tag(preprocessedTokens);

    if (m_logging) {
      LOG.info("Tweet: " + taggedTokens);
    }

    // Emit new tuples
    collector.emit(new Values(taggedTokens));
  }

}

@@ -837,16 +837,72 @@
 .bolt;%0A%0A
+import java.io.IOException;%0Aimport java.util.ArrayList;%0A
 import j
@@ -912,24 +912,24 @@
 .util.List;%0A
-
 import java.
@@ -1034,30 +1034,21 @@
 ons.
-postagger.ArkPOSTagger
+Configuration
 ;%0Aim
@@ -1395,16 +1395,187 @@
 edToken;
+%0Aimport cmu.arktweetnlp.impl.Model;%0Aimport cmu.arktweetnlp.impl.ModelSentence;%0Aimport cmu.arktweetnlp.impl.Sentence;%0Aimport cmu.arktweetnlp.impl.features.FeatureExtractor;
 %0A%0Apublic
@@ -2000,28 +2000,67 @@
 ate 
-ArkPOSTagger m_tagge
+Model m_model;%0A  private FeatureExtractor m_featureExtracto
 r;%0A%0A
@@ -2499,46 +2499,572 @@
     
-m_tagger = Ark
+try %7B%0A      String taggingModel = Configuration%0A          .get(%22global.resources.postagger.ark.model.path%22);%0A      LOG.info(%22Load ARK 
 POS
+ 
 Tagger
-.getInstance();
+ with model: %22 + taggingModel);%0A      // TODO absolute path needed for resource%0A      if ((Configuration.RUNNING_WITHIN_JAR) && (!taggingModel.startsWith(%22/%22))) %7B%0A        taggingModel = %22/%22 + taggingModel;%0A      %7D%0A      m_model = Model.loadModelFromText(taggingModel);%0A      m_featureExtractor = new FeatureExtractor(m_model, false);%0A    %7D catch (IOException e) %7B%0A      LOG.error(%22IOException: %22 + e.getMessage());%0A    %7D
 %0A  %7D
@@ -3314,17 +3314,8 @@
 s = 
-m_tagger.
 tag(
@@ -3480,10 +3480,562 @@
 );%0A  %7D%0A%0A
+  private List%3CTaggedToken%3E tag(List%3CString%3E tokens) %7B%0A    Sentence sentence = new Sentence();%0A    sentence.tokens = tokens;%0A    ModelSentence ms = new ModelSentence(sentence.T());%0A    m_featureExtractor.computeFeatures(sentence, ms);%0A    m_model.greedyDecode(ms, false);%0A%0A    List%3CTaggedToken%3E taggedTokens = new ArrayList%3CTaggedToken%3E();%0A    for (int t = 0; t %3C sentence.T(); t++) %7B%0A      TaggedToken tt = new TaggedToken(tokens.get(t),%0A          m_model.labelVocab.name(ms.labels%5Bt%5D));%0A      taggedTokens.add(tt);%0A    %7D%0A    return taggedTokens;%0A  %7D%0A
 %7D%0A

Update POS Tagger