/*
 * MiningMart Version 1.0
 * 
 * Copyright (C) 2006 Martin Scholz, Timm Euler, 
 *                    Daniel Hakenjos, Katharina Morik
 *
 * Contact: miningmart@ls8.cs.uni-dortmund.de
 *
 * A list of contributing developers (other than the copyright 
 * holders) can be found at
 * http://mmart.cs.uni-dortmund.de/downloads/download.html
 * 
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 * 
 * This program 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 General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program, see the file MM_HOME/LICENSE; if not, write
 * to the Free Software Foundation, Inc., 51 Franklin Street, Fifth
 * Floor, Boston, MA 02110-1301, USA.
 */
package edu.udo.cs.miningmart.storedProcedures;

import java.sql.SQLException;

/**
 * @author Martin Scholz
 * @version $Id: SimpleSTSP.java,v 1.4 2006/04/11 14:10:16 euler Exp $
 */
public class SimpleSTSP {

    private final DbConnector connector;
    private final Windowing   window; // for the input
    private final OutputColumnset out; // for the output
    private final double          tolerance;

    private static final int INDEXONATTRIB = 2;

    protected Windowing getWindow() {
	return this.window;
    }

    protected OutputColumnset getOutputColumnset() {
	return this.out;
    }

    protected double getTolerance() {
	return this.tolerance;
    }

    /**
     * @param dbc source of database connection if not used as stored procedure
     * @param source source table
     * @param time   time column in source table
     * @param column source column in source table
     *
     * @param target     target table to be created
     * @param timeStart  column for the start of the time intervals in the target
     * @param timeEnd    column for the end of the time intervals in the target
     * @param averageCol attribute for the average of intervals in the target
     * @param incCol     attribute for the increase within intervals in the target
     *
     * @param tolerance parameter specifying allowed deviation within one interval
    */
    public SimpleSTSP(BusinessDbConnectionSource dbc,
    		  String source, String time, String column, String target,
		      String timeStart, String timeEnd, String averageCol,
		      String incCol, double tolerance)
	throws SQLException, TimeOperatorException
	{
		this.connector = new DbConnector(dbc);
		this.tolerance = Math.abs(tolerance);
		final String timeDT =
			DataType.getColumnDataType(source, time, this.connector);
		final String columnDT =
			DataType.getColumnDataType(source, column, this.connector);

		final String numName = dbc.getNameOfNumericDatatype();
		// Input:
		final Attribute timeAttr = new Attribute(time, timeDT, numName);
		final Attribute columnAttr =
			new Attribute(column, columnDT, numName);
		if (!timeAttr.isConvertable())
			DataType.wrongDataType(target, timeAttr);
		if (!columnAttr.isConvertable())
			DataType.wrongDataType(target, columnAttr);

		// Output:
		final Attribute timeStartAttr =
			new Attribute(timeStart, timeDT, numName);
		final Attribute timeEndAttr =
			new Attribute(timeEnd, timeDT, numName);
		final Attribute averageColAttr =
			new Attribute(averageCol, columnDT, numName);
		final Attribute incColAttr =
			new Attribute(incCol, columnDT, numName);
		Attribute[] allAttributes =
			new Attribute[] {
				timeStartAttr,
				timeEndAttr,
				averageColAttr,
				incColAttr };
		this.out =
			new OutputColumnset(
				target,
				allAttributes,
				this.connector.getConnection());
		// Prepare Windowing:
		this.window = new Windowing(1, 1, this.connector);
		if (!this
			.window
			.open(
				"SELECT "
					+ columnAttr.getConvertedInput()
					+ ", "
					+ timeAttr.getConvertedInput()
					+ " FROM "
					+ source)) {
			DbConnector.infoOutput("Error while opening " + source);
			throw new TimeOperatorException(
				"Could not read from table '" + source + "'!");
		}
	}
 

        
    /**
     * Method calc() reads from the column(s) specified in the constructor
     * and writes to the specified target, using windowing.
     * <!-- Finally an index is created on the target table. -->
     * <p>
     * The calculation itself: Always starts a new interval which is
     * filled until the tolerance criterion for adding another value is
     * violated. Stops, if there is no new value for starting another
     * interval.</p>
     */
    public void calc() throws TimeOperatorException {
		TimeValue tv = nextTimeValue();
		if (tv == null) {
		    return;
		}
		while (tv != null) {
		    tv = startNewInterval(tv);
		}

		// this.getOutputColumnset().createIndexOn(INDEXONATTRIB);
		// *** No index is created in this compiler version! ***
	
		this.getWindow().close();
		this.connector.close();
    }
    

    /**
     * @param values a <i>TimeValueArray</i> object to be inserted
     * into the target table
     * */
    private void insert(TimeValueArray values) throws TimeOperatorException {
		Object[] i = new Object[] { new Long(values.getStartTime()),
				    new Long(values.getLatestTime()),
				    new Double(values.getAverageValue()),
				    new Double(values.getIncreaseByTime()) };
		this.getOutputColumnset().insert(i);
    }


    private TimeValue startNewInterval(TimeValue tv) throws TimeOperatorException
    {
		if (tv == null) {
		    return null;
		}
		TimeValueArray values = new TimeValueArray(tv);

		// Add to 'values' until end of input table or the
		// 'intoTimeValueArray' returns false:
		while ( (tv = nextTimeValue()) != null && intoTimeValueArray(values, tv) );
	
		// Then insert the interval as a new tuple into the target table:
		insert(values);
		
		// Return the TimeValue object not inserted, or 'null' if end of table:
		return tv;
    }


    /**
     * This method encapsulates the criterion for rejecting to add a
     * 'TimeValue' to an existing interval. In this case the method
     * returns 'false'.
     *
     * The criterion used: extrapolating by average increase over time
     * differs by more than 'tolerance' (parameter of the operator)
     * from the new 'TimeValue'.
    */
    private boolean intoTimeValueArray(TimeValueArray values, TimeValue tv)
    {
		double refValue = values.extrapolateValueFor(tv.getTimeAsLong());
		if (Math.abs(refValue - tv.getValue()) > this.getTolerance()) {
		    return false;
		}
		values.add(tv);
		return true;
    }


    /**
     * Reads a time-value tuple from the database.
     * @return a 'TimeValue' object representing the read time-value tuple.
     * @throws NumberFormatException if the read tuple cannot be parsed
     * */
    private TimeValue nextTimeValue() throws NumberFormatException
    {
		String[] ret = window.getNextWnd();
		if ((ret == null || ret.length < 1)) {
		    return null;
		}
		return (new TimeValue(window.getTimeFirstEntry(), ret[0]));
    }


    /** 
     * This function has to be used in the database as stored
     * procedure to calculate a simple aggregation of time series to
     * intervals.  For parameter information please refer to the
     * documentation of the constructor.
     */
    public static void dbSTSP(Object dbc,
    			  String source, String time, String column,
			      String target, String timeStart, String timeEnd,
			      String averageCol, String incCol,
			      double tolerance)
	throws SQLException, TimeOperatorException
    {
    	BusinessDbConnectionSource dbcCasted =
			(dbc instanceof BusinessDbConnectionSource) ?
				((BusinessDbConnectionSource) dbc) : null;
		
		SimpleSTSP stsp = new SimpleSTSP(dbcCasted, source, time, column, target, timeStart,
										 timeEnd, averageCol, incCol, tolerance);
		stsp.calc();
		stsp = null;
    }

}
/*
 *$Log: SimpleSTSP.java,v $
 *Revision 1.4  2006/04/11 14:10:16  euler
 *Updated license text.
 *
 *Revision 1.3  2006/04/06 16:31:16  euler
 *Prepended license remark.
 *
 *Revision 1.2  2006/03/30 16:07:15  scholz
 *fixed author tags for release
 *
 *Revision 1.1  2006/01/03 09:54:23  hakenjos
 *Initial version!
 * 
 */
