package compbio.cassandra.readers;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;

import compbio.beans.TotalByCounterBean;

public class ReaderByCounter extends CassandraReader {

	public ReaderByCounter() {
		super();
	}

	/**
	 * query protein sequences with number of jobs
	 * 
	 * @param minimalcounter
	 * 
	 * @return List<TotalByCounterBean> to the controller SequenceController
	 * 
	 **/
	public List<TotalByCounterBean> readProteinByCounter(int minimalcounter) {
		ResultSet results = CassandraQuery("SELECT Protein, JobID FROM ProteinRow;");
		return QueryByCounter(results, "Protein", minimalcounter);
	}

	/**
	 * query ip with number of jobs
	 * 
	 * @param minimalcounter
	 * 
	 * @return List<TotalByCounterBean> to the controller IPDataController
	 * 
	 **/
	public List<TotalByCounterBean> readIpByCounter(Integer minimalcounter) {
		ResultSet results = CassandraQuery("SELECT JobID, ip FROM ProteinLog;");
		return QueryByCounter(results, "ip", minimalcounter);
	}

	/**
	 * query number of jobs by counter and parameter: ip or protein
	 * 
	 * @param results
	 *            cassandra query result
	 * @param counterParameter
	 *            parameter for a query: ip or proteint
	 * @param minimalcounter
	 * 
	 * @return List<TotalByCounterBean>
	 * 
	 **/
	public List<TotalByCounterBean> QueryByCounter(ResultSet results, String counterParameter, int minimalcounter) {
		List<TotalByCounterBean> query = new ArrayList<TotalByCounterBean>();
		if (results.isExhausted())
			return null;
		List<Row> rows = results.all();
		Map<String, Integer> res = new HashMap<String, Integer>();
		for (Row r : rows) {
			String parameter = r.getString(counterParameter);
			if (res.containsKey(parameter))
				res.put(parameter, res.get(parameter) + 1);
			else
				res.put(parameter, 1);
		}
		for (Map.Entry<String, Integer> entry : res.entrySet()) {
			if (entry.getValue() > minimalcounter) {
				TotalByCounterBean bean = new TotalByCounterBean();
				bean.setTotaljobs(entry.getValue());
				bean.setName(entry.getKey());
				query.add(bean);
			}
		}
		return query;
	}

}