/**
 * Copyright (c) 2012 - 2019 Data In Motion and others.
 * All rights reserved. 
 * 
 * This program and the accompanying materials are made available under the terms of the 
 * Eclipse Public License v1.0 which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 * 
 * Contributors:
 *     Data In Motion - initial API and implementation
 */
package org.gecko.influxdb.impl;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.logging.Logger;

import org.gecko.influxdb.api.CSVReader;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.ServiceScope;

/**
 * 
 * @author ilenia
 * @since May 14, 2019
 */
@Component(scope = ServiceScope.PROTOTYPE)
public class CSVReaderImpl implements CSVReader {

	private Logger logger = Logger.getLogger(CSVReaderImpl.class.getName());	
	
	/* 
	 * (non-Javadoc)
	 * @see org.gecko.influxdb.api.CSVReader#convert(java.lang.String, java.util.Map)
	 */
	@Override	
	public Map<String, List<Object>> convert(String csvFilePath, Map<String, Object> convertOptions)
			throws IOException {
		
		InputStream inputStream = new FileInputStream(csvFilePath);
		return convert(inputStream, convertOptions);
	}

	
	/* 
	 * (non-Javadoc)
	 * @see org.gecko.influxdb.api.CSVReader#convert(java.io.InputStream, java.util.Map)
	 */
	@SuppressWarnings("unchecked")
	@Override
	public Map<String, List<Object>> convert(InputStream inputStream, Map<String, Object> convertOptions)
			throws IOException {
		
		if (convertOptions == null) {
			throw new IllegalArgumentException("The option map is mandatory");
		}
		String separator = (String) convertOptions.getOrDefault(PROP_SEPARATOR, ",");
		boolean isHeader = (boolean) convertOptions.getOrDefault(PROP_HEADER, false);
		Class<?>[] types = (Class<?>[]) convertOptions.get(PROP_TYPE_ARRAY);
		int lineNum = 0;
		Map<String, List<Object>> result = new LinkedHashMap<String, List<Object>>();		
		Scanner sc = null;
		
		try {
			sc = new Scanner(inputStream, "UTF-8");
			while (sc.hasNextLine()) {
				String line = sc.nextLine();
				String[] columns = line.split(separator);
				if (types != null && columns.length != types.length) {
					types = null;
				}
				for(int i = 0; i < columns.length; i++) {									
					
					if(isHeader) {
						if(lineNum == 0) {      
							logger.fine("col name " + columns[i]);
							result.put(columns[i], new LinkedList<Object>());
						}            			            		
						else {
							Map.Entry<String, List<Object>> element = (Map.Entry<String, List<Object>>) result.entrySet().toArray()[i];											
							Object value = columns[i];
							if (types != null) {
								value = castToType(columns[i], types[i]);
							}	
							element.getValue().add(value);
						}
					}
					else {
						if(lineNum == 0) {     
							result.put("column"+i, new LinkedList<Object>());
						}		
						Object value = columns[i];
						if (types != null) {
							value = castToType(columns[i], types[i]);
						}	
						result.get("column"+i).add(value);
					}
				} 
				lineNum++;
			}
			if (sc.ioException() != null) {
				logger.severe("Error converting CSV file " + sc.ioException());
				return null;
			}
		} finally {
			if (inputStream != null) {
				inputStream.close();
			}
			if (sc != null) {
				sc.close();
			}
		}
		return result;
	}
	
	
	/* 
	 * (non-Javadoc)
	 * @see de.dim.queueing.influxdb.api.CSVReader#convert(java.lang.String, java.lang.String, boolean)
	 */
	@Override
	public Map<String, List<Object>> convert(String csvFilePath, String separator, boolean isHeader) throws IOException {
		Map<String, Object> options = CSVReader.createDefaultOptions(separator, isHeader);
		return convert(csvFilePath, options);
	}
	
	/* 
	 * (non-Javadoc)
	 * @see org.gecko.influxdb.api.CSVReader#convert(java.io.InputStream, java.lang.String, boolean)
	 */
	@Override
	public Map<String, List<Object>> convert(InputStream inputStream, String separator, boolean isHeader)
			throws IOException {
		Map<String, Object> options = CSVReader.createDefaultOptions(separator, isHeader);
		return convert(inputStream, options);
	}

	
	

	/**
	 * @param string
	 * @return
	 */
	private Object castToType(String value, Class<?> type) {
		if (type == String.class) {
			return value;
		}
		try {
			if (type == Long.TYPE) {
				return Long.parseLong(value);
			}
			if(type == Integer.TYPE) {
				return Integer.parseInt(value);
			}
			if(type == Double.TYPE) {
				return Double.parseDouble(value);
			}
			if(type == Float.TYPE) {
				return Float.parseFloat(value);
			}
			if(type == Boolean.TYPE) {
				return Boolean.parseBoolean(value);
			}
		} catch(NumberFormatException e) {
			logger.warning("Unknown type for " + value +". Returning it as a String.");
			return value;
		}
		return null;
	}

	

}