CSVWrapper extends AbstractWrapper { @Override public boolean initialize() { setName("CSVWrapper-Thread" + (++threadCounter)); AddressBean addressBean = getActiveAddressBean(); dataFileName = addressBean.getPredicateValueWithException("file"); String csvFields = addressBean.getPredicateValueWithException("fields"); String csvFormats = addressBean.getPredicateValueWithException("formats"); 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", CSVHandler.LOCAL_TIMEZONE_ID); String nullValues = addressBean.getPredicateValueWithDefault("bad-values", ""); String strUseCounterForCheckPoint = addressBean.getPredicateValueWithDefault("use-counter-for-check-point", "false"); samplingPeriodInMsc = addressBean.getPredicateValueAsInt("sampling", 10000); samplingCountPerPeriod = addressBean.getPredicateValueAsInt("sampling-count", 250); 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"); } 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(dataFileName).getName()); if (!handler.initialize(dataFileName.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 (IOException | NumberFormatException e) { logger.error("Loading the csv-wrapper failed:" + e.getMessage(), e); return false; } dataField = handler.getDataFields(); logger.warn("Reading from: " + dataFileName); return true; }  @Override boolean initialize(); @Override void run(); @Override DataField[] getOutputFormat(); @Override String getWrapperName(); @Override void dispose(); void DEBUG_INFO(String s);  }
@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()); 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")); }
@Test public void testCheckpoints() throws IOException { String fields = "TIMED, air_temp , TIMEd , AiR_TeMp2, comments"; String formats = "Timestamp(d.M.y ) , Numeric , timestamp(k:m) , numeric ,String "; String data = "01.01.2009,1,10:10,10,\"Ali Salehi\"\n"+ "01.01.2009,2,10:11,11,\"Ali Salehi\"\n"+ "01.01.2009,3,10:12,12,\"Ali Salehi\"\n"; CSVHandler wrapper = new CSVHandler(); assertEquals(true,wrapper.initialize("test.csv.csv", fields,formats,',','\"',0,"NaN,-1234,4321")); ArrayList<TreeMap<String, Serializable>> parsed = wrapper.parseValues(new StringReader(data), -1, 250); assertEquals(3, parsed.size()); assertEquals(wrapper.work(new StringReader(data), CHECK_POINT_DIR, 250).size(), parsed.size()); assertEquals(true,((Long)parsed.get(0).get("timed"))<((Long)parsed.get(1).get("timed"))); long recentTimestamp = ((Long)parsed.get(parsed.size()-1).get("timed")); data+="01.01.2009,3,10:12,12,\"Ali Salehi\"\n"; assertEquals(0,wrapper.parseValues(new StringReader(data), recentTimestamp, 250).size()); assertEquals(0,wrapper.work(new StringReader(data), CHECK_POINT_DIR, 250).size()); data+="01.01.2009,3,10:12,12,\"Ali Salehi\"\n"; data+="01.01.2009,3,10:11,12,\"Ali Salehi\"\n"; data+="01.01.2009,3,10:10,12,\"Ali Salehi\"\n"; assertEquals(0,wrapper.parseValues(new StringReader(data), recentTimestamp, 250).size()); assertEquals(0,wrapper.work(new StringReader(data), CHECK_POINT_DIR, 250).size()); data+="01.01.2009,3,10:13,13,\"Ali Salehi\"\n"; assertEquals(1,wrapper.parseValues(new StringReader(data), recentTimestamp, 250).size()); assertEquals(1,wrapper.work(new StringReader(data), CHECK_POINT_DIR, 250).size()); data="###########################\n\n\n\n,,,,,,,,,\n\n\n"; wrapper.setSkipFirstXLines(1); assertEquals(0,wrapper.parseValues(new StringReader(data), recentTimestamp, 250).size()); assertEquals(0,wrapper.work(new StringReader(data), CHECK_POINT_DIR, 250).size()); }