analysis/org.eclipse.stem.analysis/src/org/eclipse/stem/analysis/util/CSVScenarioStreamer.java

package org.eclipse.stem.analysis.util;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import org.eclipse.stem.analysis.ScenarioInitializationException;

/*******************************************************************************
 * Copyright (c) 2007, 2008 IBM Corporation 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:
 *     IBM Corporation - initial API and implementation
 *******************************************************************************/


/**
 * Adds methods for streaming data instead of loading all data files into memory
 */
public class CSVScenarioStreamer extends CSVscenarioLoader {

	public static final int NUM_LEAD_COLS = 2;
	
	Map<Integer, Map<String, BufferedReader>> streamReaders = new HashMap<Integer, Map<String, BufferedReader>>();
	Map<Integer, Map<String, Integer>> streamReaderCurrentRow = new HashMap<Integer, Map<String, Integer>>();
	Map<Integer, Map<String, List<Double>>> streamReadersLastRow = new HashMap<Integer, Map<String, List<Double>>>();
	
	Map<Integer, List<String>> headersMap = new HashMap<Integer, List<String>>();
	
	private int row = -1; // Which row
	
	/**
	 * @param path
	 */
	
	public CSVScenarioStreamer(String path) throws ScenarioInitializationException {
		super(path);
	}
	
	
	/**
	 * prepareForStreaming
	 * 
	 * @param resolution The desired resolution, or -1 of all files should be streamed
	 */
	
	public void prepareForStreaming(int resolution) throws IOException {
		
		for(int i = 0; i < diseaseData.size(); i ++) {
			File f = diseaseData.get(i);
			String name = f.getName();
			if(f.getName().startsWith(".")) continue;//$NON-NLS-1$
			int _idx = name.lastIndexOf('_');
			int dotidx = name.lastIndexOf('.');
			if(_idx <0 || dotidx < 0 || dotidx < _idx) continue; // not a data file
			String state = name.substring(0, _idx);

			// discovered a state
			registerKey(state);
			
			int res = Integer.parseInt(name.substring(_idx+1, dotidx));
			if(resolution != -1 && res != resolution) continue; // wrong resolution
			openFile(state, f, res);
		}
	}
	
	/**
	 * openFile. Open data file and prepare for streaming
	 * 
	 * @param state The state
	 * @param File file
	 * @param adminLevel
	 */
	
	public void openFile(String state, File f, int adminLevel) throws IOException {
		BufferedReader reader = new BufferedReader(new FileReader(f));
		
		if(!streamReaders.containsKey(adminLevel)) {
			streamReaders.put(adminLevel, new HashMap<String, BufferedReader>());
			streamReaderCurrentRow.put(adminLevel, new HashMap<String, Integer>());
			streamReadersLastRow.put(adminLevel, new HashMap<String, List<Double>>());
		}
		Map<String, BufferedReader>stateReaderMap = streamReaders.get(adminLevel);
		Map<String, Integer>stateRowMap = streamReaderCurrentRow.get(adminLevel);
		Map<String, List<Double>>stateRowData = streamReadersLastRow.get(adminLevel);
		stateReaderMap.put(state, reader);
		stateRowMap.put(state, -1); // no row read yet, first row is row 0
		stateRowData.put(state, null); // no data yet
		String header;
		
		if ( (header=reader.readLine()) != null ) { 
			if(!headersMap.containsKey(adminLevel)) {
				List<String>headerElements = new ArrayList<String>();
				headersMap.put(adminLevel, headerElements);
				StringTokenizer st = new StringTokenizer(header, ",");//$NON-NLS-1$
	
				int col = 0;
				while (st.hasMoreTokens()) {
					String val = st.nextToken().trim();
					if(col < NUM_LEAD_COLS) {++col;continue;}
					headerElements.add(val);
					++col;
				}
			}
		} // read the header
		/////////////////////
	}
	
	/**
	 * streamRow Stream next row of data
	 * 
	 * 
	 * @param adminLevel which admin level
	 * @param state Which state (e.g. S)
	 * @return List<Double> The data, or null if end of line is encountered;
	 */
	
	public List<Double>streamRow(int adminLevel, String state) throws IOException {
		BufferedReader br = streamReaders.get(adminLevel).get(state);
		if(br == null) return null; // stream is closed
		String data;
		if ( (data=br.readLine()) != null ) { 
			
			if(streamReadersLastRow.get(adminLevel).get(state) == null) 
				streamReadersLastRow.get(adminLevel).put(state, new ArrayList<Double>());
			List<Double>result = streamReadersLastRow.get(adminLevel).get(state);
			result.clear();
			StringTokenizer st = new StringTokenizer(data, ",");//$NON-NLS-1$
			
			int col = 0;
			while (st.hasMoreTokens()) {
				String val = st.nextToken().trim();
				if(col < NUM_LEAD_COLS) {++col;continue;}
				result.add(Double.parseDouble(val));
				++col;
			}
			streamReaderCurrentRow.get(adminLevel).put(state, streamReaderCurrentRow.get(adminLevel).get(state)+1);
			
			return result;
		}
		br.close();
		streamReaders.get(adminLevel).put(state, null);// gc reader
		return null; // end of file reached
	}
	
	public int getCurrentRow(int adminLevel, String state) {
		return streamReaderCurrentRow.get(adminLevel).get(state);
	}
	
	/**
	 * streamRow Stream next row of data
	 * 
	 * 
	 * @param adminLevel which admin level
	 * @param state Which state (e.g. S)
	 * @return List<Double> The data
	 */
	
	public List<Double>getLastFetchedRow(int adminLevel, String state) {
		List<Double>result = streamReadersLastRow.get(adminLevel).get(state);
		return result;
	}

	public int getDataPosition(int adminLevel, String key) {
		return headersMap.get(adminLevel).indexOf(key);
	}
}