class CSVWrapper extends AbstractWrapper {

    public boolean initialize() {
        AddressBean addressBean = getActiveAddressBean();
        dataFile = addressBean.getPredicateValueWithException("file");
        String csvFields = addressBean.getPredicateValueWithException("fields");
        String csvFormats = addressBean.getPredicateValueWithException("formats");
        //String csvSeparator = addressBean.getPredicateValueWithDefault("separator",",");
        String value = addressBean.getPredicateValue("separator");
        String csvSeparator = (value == null || value.length() == 0) ? "," : value;
        checkPointDir = addressBean.getPredicateValueWithDefault("check-point-directory", "./csv-check-points");
        String csvStringQuote = addressBean.getPredicateValueWithDefault("quote", "\"");
        int skipFirstXLine = addressBean.getPredicateValueAsInt("skip-first-lines", 0);
        String timezone = addressBean.getPredicateValueWithDefault("timezone", handler.LOCAL_TIMEZONE_ID);
        String nullValues = addressBean.getPredicateValueWithDefault("bad-values", "");
        String strUseCounterForCheckPoint = addressBean.getPredicateValueWithDefault("use-counter-for-check-point", "false");
        samplingPeriodInMsc = addressBean.getPredicateValueAsInt("sampling", 10000);

        /*
        DEBUG_INFO(dataFile);
        */

        if (csvSeparator != null && csvSeparator.length() != 1) {
            logger.warn("The provided CSV separator:>" + csvSeparator + "< should only have  1 character, thus ignored and instead \",\" is used.");
            csvSeparator = ",";
        }

        if (csvStringQuote.length() != 1) {
            logger.warn("The provided CSV quote:>" + csvSeparator + "< should only have 1 character, thus ignored and instead '\"' is used.");
            csvStringQuote = "\"";
        }

        try {
            if (strUseCounterForCheckPoint.equalsIgnoreCase("true")) {
                useCounterForCheckPoint = true;
                logger.warn("Using counter-based check points");
            }
            //String checkPointFile = new File(checkPointDir).getAbsolutePath()+"/"+(new File(dataFile).getName())+"-"+addressBean.hashCode();
            StringBuilder checkPointFile = new StringBuilder()
                    .append(new File(checkPointDir).getAbsolutePath())
                    .append("/")
                    .append(addressBean.getVirtualSensorName())
                    .append("_")
                    .append(addressBean.getInputStreamName())
                    .append("_")
                    .append(addressBean.getWrapper())
                    .append("_")
                    .append(new File(dataFile).getName());
            if (!handler.initialize(dataFile.trim(), csvFields, csvFormats, csvSeparator.toCharArray()[0], csvStringQuote.toCharArray()[0], skipFirstXLine, nullValues, timezone, checkPointFile.toString()))
                return false;

            String val = FileUtils.readFileToString(new File(checkPointFile.toString()), "UTF-8");
            long lastItem = 0;
            if (val != null && val.trim().length() > 0)
                lastItem = Long.parseLong(val.trim());
            logger.warn("Latest item: "+lastItem);

            if (useCounterForCheckPoint) {
                processedLineCounter = lastItem;
            }

        } catch (Exception e) {
            logger.error("Loading the csv-wrapper failed:" + e.getMessage(), e);
            return false;
        }

        dataField = handler.getDataFields();

        logger.warn("Reading from: " + dataFile);

        return true;
    }

}

class CSVWrapperTest {

	private final String CSV_FILE_NAME;
	private final String CHECK_POINT_DIR;

	@Test
	public void testFieldConverting() throws IOException {

		String fields = "TIMED, air_temp , TIMED , AiR_TeMp2";
		String formats = "Timestamp(d.M.y ) , Numeric , timestamp(k:m) , numeric    ";
		String badFormat = "Timestamp(d.M.y k:m) , numeric , numeric, numeric,numeric,dollluble ";
		String badFormat2 ="Timestamp(d.Mjo0o.y k:m) , numeric, numeric, numeric";
		
		CSVHandler wrapper = new CSVHandler();
		assertEquals(false,wrapper.initialize("test.csv.csv", fields,badFormat,',','\"',0,"NaN,-1234,4321"));
		assertEquals(false,wrapper.initialize("test.csv.csv", fields,badFormat,',','\"',0,"NaN,-1234,4321"));
		assertEquals(false,wrapper.initialize("test.csv.csv", fields,badFormat2,',','\"',0,"NaN,-1234,4321"));
		
		assertEquals(true,wrapper.initialize("test.csv.csv", fields,formats,',','\"',0,"NaN,-1234,4321"));
		
		FileUtils.writeStringToFile(new File(wrapper.getCheckPointFile()),  "","UTF-8");
		String[] formatsParsed = wrapper.getFormats();
		String[] fieldsParsed =  wrapper.getFields();
		assertEquals(true,compare(fieldsParsed, new String[] {"timed","air_temp","timed","air_temp2"}));
		assertEquals(true,compare(formatsParsed, new String[] {"Timestamp(d.M.y )","Numeric","timestamp(k:m)","numeric"}));
		
		TreeMap<String, Serializable> se = wrapper.convertTo(wrapper.getFormats(),wrapper.getFields(),wrapper.getNulls(),new String[] {} , wrapper.getSeparator());
		assertEquals(wrapper.getFields().length-1, se.keySet().size());//timestamp is douplicated.
		assertEquals(null, se.get("timed"));
		se = wrapper.convertTo(wrapper.getFormats(),wrapper.getFields(),wrapper.getNulls(),new String[] {"","","","-1234","4321","NaN"} , wrapper.getSeparator());
		assertEquals(null, se.get("timed"));
		
		se = wrapper.convertTo(wrapper.getFormats(),wrapper.getFields(),wrapper.getNulls(),new String[] {"","","","-1234","4321","NaN"} , wrapper.getSeparator());
		assertEquals(null, se.get("timed"));
		
		se = wrapper.convertTo(wrapper.getFormats(),wrapper.getFields(),wrapper.getNulls(),new String[] {"01.01.2009","1234","","-4321","ignore-me","NaN"} , wrapper.getSeparator());
		long parsedTimestamp = (Long)se.get("timed");
		assertEquals(true,parsedTimestamp>0);
		assertEquals(1234.0, se.get("air_temp"));
		assertEquals(-4321.0, se.get("air_temp2"));
		
		se = wrapper.convertTo(wrapper.getFormats(),wrapper.getFields(),wrapper.getNulls(),new String[] {"01.01.2009","-1234","10:10","-4321","ignore-me","NaN"} , wrapper.getSeparator());
		assertEquals(true,((Long)se.get("timed"))>parsedTimestamp);
		assertNull(se.get("air_temp"));
	
	}

}