/* $Id: TspCdfWriter.java,v 1.3 2005/10/05 00:11:47 sgalles Exp $
 * -----------------------------------------------------------------------
 * 
 * TSP Library - core components for a generic Transport Sampling Protocol.
 * 
 * Copyright (c) 2002 Yves DUFRENNE, Stephane GALLES, Eric NOULARD and Robert PAGNOT 
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 * 
 * -----------------------------------------------------------------------
 * 
 * Project    : TSP
 * SubProject : jTSP
 * Maintainer : tsp@astrium-space.com
 * Component  : Consumer
 * 
 * -----------------------------------------------------------------------
 * 
 * Purpose   : Writer of a TSP stream in CDF format
 * 
 * -----------------------------------------------------------------------
 */

package tsp.consumer.app.jcdfwriter;

import gsfc.nssdc.cdf.CDF;
import gsfc.nssdc.cdf.CDFException;
import gsfc.nssdc.cdf.Variable;

import java.io.File;
import java.util.HashMap;
import java.util.Map;

import tsp.consumer.util.aggregator.TspAggrConsumerFactory;
import tsp.consumer.util.aggregator.TspAggrURL;
import tsp.consumer.util.configuration.SampleConfigXMLMapper;
import tsp.consumer.util.configuration.SpecSamplingConfig;
import tsp.consumer.util.configuration.decorator.AggrSamplingConfig;
import tsp.consumer.util.configuration.decorator.RenamedSimpleSampleSymbolInfo;
import tsp.core.common.url.TspURLException;
import tsp.core.consumer.TspConsumerException;
import tsp.core.consumer.simple.TspSimpleConsumer;
import tsp.core.consumer.simple.TspSimpleRequestSample;
import tsp.core.consumer.simple.TspSimpleSample;
import tsp.core.consumer.simple.TspSimpleSampleSymbolInfo;
import tsp.core.consumer.simple.TspSimpleSession;
import gsfc.nssdc.cdf.*;

class TspCdfWriter {
	
	private String confSamplingPath;
	private String outputFile;
	private long samplingDuration;
	private long startSystemTime;
	private SpecSamplingConfig samplingConfig;
	private TspAggrConsumerFactory aggrConsumerFactory;
	CDF cdf;
	
	private static void printUsage(){
		System.out.println("Usage : [classpath:]samplingXmlConfPath CdfOutputFilePath samplingSecondDuration");
	}
	
	public static void main(String[] args) throws Exception{
					
		if(args.length < 3 || args.length > 3){
			printUsage();
			System.exit(1);
		}
		Integer samplingDuration;
		try{
			samplingDuration = new Integer(args[2]);
		}catch(NumberFormatException e){
			printUsage();
			throw e;
		}
		
		TspCdfWriter writer = new TspCdfWriter(args[0],args[1],samplingDuration.longValue(), args);
		writer.sample();				
	}
	
	/**
	 * @param confSamplingPath
	 * @param outputFile
	 * @param nbSamples
	 */
	public TspCdfWriter(String confSamplingPath, String outputFile,
			long samplingDuration, String args[]) {
		super();
		this.confSamplingPath = confSamplingPath;
		this.outputFile = outputFile;
		this.samplingDuration = samplingDuration*1000;
		this.aggrConsumerFactory = new TspAggrConsumerFactory(args);
		
	}
	
		
	public void sample() throws TspCdfWriterException {
		
		try{
			// read aggregated config sampling
			SampleConfigXMLMapper parser = new SampleConfigXMLMapper();
			System.out.println("Using sampling confguration : " + confSamplingPath);
			SpecSamplingConfig specConfig  = parser.createSpecSamplingConfig(confSamplingPath);
			AggrSamplingConfig aggrConfig = new AggrSamplingConfig(specConfig);
			
			// open aggregated consumers
			TspAggrURL[] urls = aggrConfig.getAllAggrURL();
			TspSimpleConsumer consumer = aggrConsumerFactory.createAggrConsumer(urls);
			TspSimpleSession session = consumer.openSession();
			
			// prepare sampling for the configured symbols
			RenamedSimpleSampleSymbolInfo[] renamedInfos = aggrConfig.getAllRenamedSymbols();
			TspSimpleSampleSymbolInfo[] sampleInfos = RenamedSimpleSampleSymbolInfo.toSimpleSampleSymbolInfoArray(renamedInfos);			
			TspSimpleSampleSymbolInfo[] answerInfos = (session.requestSample(new TspSimpleRequestSample(sampleInfos))).getSymbolInfos();			
			
			// prepare CDF file opening
			File cdfFile = new File(outputFile);
			System.out.println("Sampling to file " + cdfFile.getName());
			if (cdfFile.exists()) {
				throw new TspCdfWriterException("File " + cdfFile.getAbsolutePath() + " already exists");
			}			
			
			// open cdf file;
			cdf  = CDF.create(cdfFile.getAbsolutePath());
			// create CDF variables
			Map bundle = createCdfVar(answerInfos, renamedInfos);			
			// set CDF file parameters
			setCdfConfig();
			
			/* begin sampling */
			session.requestSampleInit();
								
			System.out.println("Waiting for first element...");			
			session.waitForData();
			startSystemTime = System.currentTimeMillis();
			
			System.out.println("Sampling for " + samplingDuration/1000 + " seconds");
			TspSimpleSample sample = new TspSimpleSample();
			long count = 0;			
			while (System.currentTimeMillis() - startSystemTime < samplingDuration) {
				session.waitForData();				
				session.fetchSample(sample);				
				CdfVarBundle var = (CdfVarBundle)bundle.get(new Integer(sample.getIndex()));
				// Add sample
				var.getVar().putScalarData(var.getNextPos(), new Double(sample.getValue()));
				if((count++)%1000 == 0) {System.out.print(".");} 
			}
			/* end sampling */
			session.requestSampleFinalize();
			/* close Session */
			session.close();
			// close cdf file
			cdf.close();
			System.out.println("");
			System.out.print("Done !");
			
		}
		catch (TspConsumerException e) {
			throw new TspCdfWriterException(e);
		} catch (TspURLException e) {
			throw new TspCdfWriterException(e);
		} catch (Exception e) {
			throw new TspCdfWriterException(e);
		}		
		
	}
	
	
	private Map createCdfVar(TspSimpleSampleSymbolInfo[] infos, RenamedSimpleSampleSymbolInfo[] renamed) throws CDFException{
		Map cdfVar = new HashMap();
		for(int i = 0 ; i < infos.length ; i++){
			String name = renamed[i].getRenamed();
			Variable v = Variable.create(cdf, name, CDFConstants.CDF_DOUBLE, 
					1, 0, new long [] {1},
					CDFConstants.VARY, 
					new long [] {CDFConstants.NOVARY});
			v.setSparseRecords(CDFConstants.PREV_SPARSERECORDS);
			CdfVarBundle bundle = new CdfVarBundle(v,infos[i].getPeriod());
			cdfVar.put(new Integer(infos[i].getIndex()), bundle);
		}		
		return cdfVar;
		
	}
	
	private void setCdfConfig() throws CDFException{
		cdf.setMajority(CDFConstants.ROW_MAJOR);         /* Default is ROW_MAJOR */			
		cdf.selectNegtoPosfp0(CDFConstants.NEGtoPOSfp0off);
		cdf.selectCDFCacheSize(400L);            
		cdf.selectCompressCacheSize(500L);
		cdf.selectStageCacheSize(600L);
	}
	
	private static class CdfVarBundle{
		private Variable var;
		private long pos;
		private int period;
		
		
		public CdfVarBundle(Variable var, int period) {
			super();
			this.var = var;
			this.pos = 0 - period;
			this.period = period;
		}
		
		public long getNextPos() {
			return (pos+=period);
		}
		
		public Variable getVar() {
			return var;
		}
	}	
		
}
