Repository /Debellor/debellor-1.0.jar:org.debellor.base.TimeWindows

/*
 *  Debellor
 *
 *  Copyright (C) 2008-2009 by Marcin Wojnarski
 *
 *  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 3 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.  If not, see http://www.gnu.org/licenses/.
 */
package org.debellor.base;

import java.util.Arrays;

import org.debellor.core.Cell;
import org.debellor.core.DataObject;
import org.debellor.core.DataType;
import org.debellor.core.Sample;
import org.debellor.core.Sample.SampleType;
import org.debellor.core.data.DataVector;
import org.debellor.core.data.DataVector.DataVectorType;
import org.debellor.core.exception.parameters.IllegalParameterValueException;

/**
 * Treats input samples as elements of a time series
 * and generates time windows composed of a fixed number of
 * consecutive input samples, with every possible offset from the
 * beginning of the input stream.
 * Offsets are scanned in increasing order.
 * 
 * <p> Output samples have the type of a {@link DataVector},
 * whose elements are the whole input samples
 * (actually the data field of input samples).
 * 
 * <p> Decisions of output samples are <code>null</code>.
 * 
 * @author Marcin Wojnarski
 *
 */
public class TimeWindows extends Cell {

	private int length = 3;

	/** Cyclic buffer of length <code>length</code> */
	private DataObject[] items;
	
	/** Temporary variable used in onNext */
	private DataObject[] tempItems;
	
	/** Position in <code>samples</code> where the next input sample will be stored. */
	private int pos;

	private Stream input;
	
	
	public TimeWindows() {
		super(false);
	}

	@Override
	protected SampleType onOpen() throws Exception {
		length = parameters.getAsInt("windowLength");
		if(length < 1) throw new IllegalParameterValueException("windowLength", parameters);
		
		input = openInputStream();
		SampleType inputType = input.sampleType;
		
		DataType[] itemTypes = new DataType[length];
		Arrays.fill(itemTypes, inputType.data);
		DataVectorType dataType = new DataVectorType(itemTypes);
		SampleType type = new SampleType(dataType, DataType.NOT_PRESENT);
		
		items = new DataObject[length];
		tempItems = new DataObject[length];
		pos = 0;
		
		// fill 'samples' with the first (length-1) input samples
		for(; pos < length-1; pos++) {
			Sample s = input.next();
			if(s == null) return null;			
			items[pos] = s.data;
		}

		return type;
	}

	@Override
	protected Sample onNext() throws Exception {
		Sample s = input.next();
		if(s == null) return null;
		
		items[pos++] = s.data;
		if(pos == length)
			pos = 0;
		
		// "unroll" the cyclic list of items
		for(int i = 0, j = pos; i < length; i++, j++) {
			if(j == length)  j = 0;
			tempItems[i] = items[j];
		}
		DataVector window = new DataVector(tempItems);
		
		return new Sample(window, null);
	}

	@Override
	protected void onClose() throws Exception {
		input.close();
		items = null;
		tempItems = null;
	}

}