// Path: src/main/java/eu/socialsensor/focused/crawler/models/ImageVector.java
// public class ImageVector {
// 	public String id;
// 	public double[] v;
// 	public String url;
// 
// 	public ImageVector(String id, String url, double[] v) {
// 		this.id = id;
// 		this.url = url;
// 		this.v = v;
// 	}
// }


import static backtype.storm.utils.Utils.tuple;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.log4j.Logger;
import com.mathworks.toolbox.javabuilder.MWException;
import eu.socialsensor.focused.crawler.models.ImageVector;
import eu.socialsensor.framework.common.domain.Concept;
import eu.socialsensor.framework.common.domain.Concept.ConceptType;
import eu.socialsensor.framework.common.domain.MediaItem;
import gr.iti.mklab.detector.examples.Example;
import gr.iti.mklab.detector.examples.ReadFileToStringList;
import gr.iti.mklab.detector.smal.ConceptDetector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
		
		_logger = Logger.getLogger(ConceptDetectionBolt.class);
		
		_collector = collector;
		
		queue = new LinkedBlockingQueue<Pair<ImageVector, MediaItem>>();
		try {
			
			_detector = new ConceptDetector(matlabFile);
			
			conceptDetectionTask = new ConceptDetectionTask(queue);
			conceptDetectionThread = new Thread(conceptDetectionTask);
			conceptDetectionThread.start();
		}
		catch(Exception e) {
			_logger.fatal(e);
		}
	}

	public void execute(Tuple input) {
		try {
			
			ImageVector imgVec = (ImageVector) input.getValueByField("ImageVector");
			MediaItem mediaItem = (MediaItem) input.getValueByField("MediaItem");
			
			if(imgVec == null) {
				synchronized(_collector) {
					_collector.emit(tuple(mediaItem));
				}
			}
// Path: src/main/java/eu/socialsensor/focused/crawler/models/ImageVector.java
// public class ImageVector {
// 	public String id;
// 	public double[] v;
// 	public String url;
// 
// 	public ImageVector(String id, String url, double[] v) {
// 		this.id = id;
// 		this.url = url;
// 		this.v = v;
// 	}
// }


// Path: src/main/java/eu/socialsensor/focused/crawler/bolts/media/ConceptDetectionBolt.java
import static backtype.storm.utils.Utils.tuple;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.log4j.Logger;
import com.mathworks.toolbox.javabuilder.MWException;
import eu.socialsensor.focused.crawler.models.ImageVector;
import eu.socialsensor.framework.common.domain.Concept;
import eu.socialsensor.framework.common.domain.Concept.ConceptType;
import eu.socialsensor.framework.common.domain.MediaItem;
import gr.iti.mklab.detector.examples.Example;
import gr.iti.mklab.detector.examples.ReadFileToStringList;
import gr.iti.mklab.detector.smal.ConceptDetector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

		
		_logger = Logger.getLogger(ConceptDetectionBolt.class);
		
		_collector = collector;
		
		queue = new LinkedBlockingQueue<Pair<ImageVector, MediaItem>>();
		try {
			
			_detector = new ConceptDetector(matlabFile);
			
			conceptDetectionTask = new ConceptDetectionTask(queue);
			conceptDetectionThread = new Thread(conceptDetectionTask);
			conceptDetectionThread.start();
		}
		catch(Exception e) {
			_logger.fatal(e);
		}
	}

	public void execute(Tuple input) {
		try {
			
			ImageVector imgVec = (ImageVector) input.getValueByField("ImageVector");
			MediaItem mediaItem = (MediaItem) input.getValueByField("MediaItem");
			
			if(imgVec == null) {
				synchronized(_collector) {
					_collector.emit(tuple(mediaItem));
				}
			}
			else {