@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")); }
public boolean initialize() { AddressBean addressBean = getActiveAddressBean(); dataFile = 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", 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); 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(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; }
CSVWrapper extends AbstractWrapper { public boolean initialize() { AddressBean addressBean = getActiveAddressBean(); dataFile = 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", 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); 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(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; } }
CSVWrapper extends AbstractWrapper { public boolean initialize() { AddressBean addressBean = getActiveAddressBean(); dataFile = 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", 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); 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(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; }  }
CSVWrapper extends AbstractWrapper { public boolean initialize() { AddressBean addressBean = getActiveAddressBean(); dataFile = 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", 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); 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(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; }  boolean initialize(); void run(); DataField[] getOutputFormat(); String getWrapperName(); void dispose(); void DEBUG_INFO(String s); }
CSVWrapper extends AbstractWrapper { public boolean initialize() { AddressBean addressBean = getActiveAddressBean(); dataFile = 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", 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); 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(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; }  boolean initialize(); void run(); DataField[] getOutputFormat(); String getWrapperName(); void dispose(); void DEBUG_INFO(String s);  }