<commit_before>'use strict';

// 
// Base class for series classes.
//

var assert = require('chai').assert; 
var E = require('linq');
var moment = require('moment');
var ArrayIterator = require('./iterators/array');
var validateIterator = require('./iterators/validate');
var SkipIterator = require('./iterators/skip');
var SkipWhileIterator = require('./iterators/skip-while');
var TakeIterator = require('../src/iterators/take');
var TakeWhileIterator = require('../src/iterators/take-while');
var SelectIterator = require('../src/iterators/select');
var SelectManyIterator = require('../src/iterators/select-many');
var PairIterator = require('../src/iterators/pair');
var WhereIterator = require('../src/iterators/where');
var CountIterator = require('../src/iterators/count');
var EmptyIterator = require('../src/iterators/empty');
var PairsIterable = require('../src/iterables/pairs');
var SelectIterable = require('../src/iterables/select');
var ArrayIterable = require('../src/iterables/array');
var extend = require('extend');

//
// Represents a time series.
//
var Series = function (config) {

	var self = this;

	if (!self.Constructor) {
		self.Constructor = Series;
	}

	if (!config) {
		self.__iterable = {
			getIterator: function () {
				return new EmptyIterator();
			},
		};
		return;
	}

	assert.isObject(config, "Expected 'config' parameter to Series constructor to be an object with options for initialisation.");

	if (config.__iterable) {
		// Setting the inner iterable directly.
		// Power to you!
		self.__iterable = config.__iterable;
		return;
	}

	if (config.iterable) {

		if (Object.isFunction(config.iterable)) {
			self.__iterable = {
				getIterator: config.iterable
			};
		}
		else {
			assert.isObject(config.iterable, "Expected 'iterable' field of 'config' parameter to Series constructor to either be a function that returns a pairs iterator or an object with a getIterator function.");
			assert.isFunction(config.iterable.getIterator, "Expected 'iterable' field of 'config' parameter to Series constructor to either be a function that returns a pairs iterator or an object with a getIterator function.");

			self.__iterable = config.iterable; 
		}
		return;
	}

	if (config.values) {
		if (!Object.isFunction(config.values)) {
			assert.isArray(config.values, "Expected 'values' field of 'config' parameter to Series constructor be an array of values or a function that returns an iterator.");
		}
	}

	if (config.index) {
		if (!Object.isFunction(config.index) && !Object.isArray(config.index)) {
			assert.isObject(config.index, "Expected 'index' field of 'config' parameter to Series constructor to be an array, function or Series object.");
		}
	}

	var indexIterable = null;
	var valuesIterable = null;

	var index = config.index;
	if (!index) {
		indexIterable = {
			getIterator: function () {
				return new CountIterator();
			},			
		};
	}
	else if (Object.isFunction(index)) {
		indexIterable = {
			getIterator: index,
		};
	}
	else if (Object.isArray(index)) {
		indexIterable = {
			getIterator: function () {
				return new ArrayIterator(index);
			},
		};
	}
	else {		
		indexIterable = {
			getIterator: function () {
				return new SelectIterator(
					index.getIterator(),
					function (pair) {
						return pair[1]; // Extract the value from the index.
					}
				)

			},
		};
	}

	var values = config.values;
	if (!values) {
		valuesIterable = {
			getIterator: function () {
				return new EmptyIterator();
			},			
		};
	}
	else if (Object.isFunction(values)) {
		valuesIterable = {
			getIterator: values,
		};
	}
	else {
		valuesIterable = {
			getIterator: function () {
				return new ArrayIterator(values);
			},
		};
	}

	self.__iterable = new PairsIterable(indexIterable, valuesIterable);
};

module.exports = Series;

var concatSeries = require('./concat-series');
var DataFrame = require('./dataframe');
var zipSeries = require('./zip-series');

/**
 * Get an iterator for index & values of the series.
 */
Series.prototype.getIterator = function () {
	var self = this;
	return self.__iterable.getIterator();
};

/**
 * Retreive the index of the series.
 */
Series.prototype.getIndex = function () {
	var self = this;
	return new Series({
		values: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					return pair[0]; // Extract the index.
				}
			);
		},
	});
};

/**
 * Apply a new index to the Series.
 * 
 * @param {array|Series} newIndex - The new index to apply to the Series.
 */
Series.prototype.withIndex = function (newIndex) {

	if (!Object.isArray(newIndex)) {
		assert.isObject(newIndex, "'Expected 'newIndex' parameter to 'Series.withIndex' to be an array or Series object.");
	}	

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				var indexIterator = Object.isArray(newIndex)
					? new ArrayIterator(newIndex)
					: new SelectIterator(
						newIndex.getIterator(),
						function (pair) {
							return pair[1]; // Extract index value.
						}
					)
					;
				return new PairIterator(
					indexIterator,
					new SelectIterator(
						self.getIterator(),
						function (pair) {
							return pair[1]; // Extract value.
						}
					)
				);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		}
	});
};

/**
 * Reset the index of the data frame back to the default sequential integer index.
 */
Series.prototype.resetIndex = function () {

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new PairIterator(
					new CountIterator(), // Reset index.
					new SelectIterator(
						self.getIterator(),
						function (pair) {
							return pair[1]; // Extract value.
						}
					)
				)
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		},
	});
};

/**
 * Skip a number of rows in the series.
 *
 * @param {int} numRows - Number of rows to skip.
 */
Series.prototype.skip = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'skip' function to be a number.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new SkipIterator(self.getIterator(), numRows);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		}
	}); 	
};

/**
 * Skips values in the series while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.skipWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new self.Constructor({
		__iterable: {

			getIterator: function () {
				return new SkipWhileIterator(
					self.getIterator(),
					function (pair) {
						return predicate(pair[1]); // Extract the value.
					}
				); 
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		}
	}); 	
};

/**
 * Skips values in the series until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.skipUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.skipWhile(function (value) { 
		return !predicate(value); 
	});
};

/**
 * Take a number of rows in the series.
 *
 * @param {int} numRows - Number of rows to take.
 */
Series.prototype.take = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'take' function to be a number.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new TakeIterator(self.getIterator(), numRows);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		},
	});
};

/**
 * Take values from the series while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.takeWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new TakeWhileIterator(
					self.getIterator(),
					function (pair) {
						return predicate(pair[1]); // Extract the value.
					}
				); 
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},

		},
	}); 	
};

/**
 * Take values from the series until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.takeUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.takeWhile(function (value) { 
		return !predicate(value); 
	});
};

/**
 * Filter a series by a predicate selector.
 *
 * @param {function} filterSelectorPredicate - Predicte function to filter rows of the series.
 */
Series.prototype.where = function (filterSelectorPredicate) {
	assert.isFunction(filterSelectorPredicate, "Expected 'filterSelectorPredicate' parameter to 'where' function to be a function.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new WhereIterator(
					self.getIterator(), 
					function (pair) {
						return filterSelectorPredicate(pair[1]); // Extract the value.
					}
				);
			},

			getColumnNames: function () {
				return self.getColumnNames(); // Doesn't change column names, or order.
			}
		},
	}); 	
};


/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each value to create a new series.
 */
Series.prototype.select = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'select' function to be a function.");

	var self = this;
	return new self.Constructor({
		__iterable: new SelectIterable(self, selector),
	}); 	
};

/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each index/value to a create a new series.
 */
Series.prototype.selectPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'selectPairs' function to be a function.");

	var self = this;
	return new self.Constructor({
		iterable: function () {
			return new SelectIterator(self.getIterator(), 
				function (pair) {
					var newPair = selector(pair[1], pair[0]);
					if (!Object.isArray(newPair) || newPair.length !== 2) {
						throw new Error("Expected return value from 'Series.selectPairs' selector to be a pair, that is an array with two items: [index, value].");
					}
					return newPair;
				}
			);
		},		
	}); 	
};

/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each value to a different data structure.
 */
Series.prototype.selectMany = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'Series.selectMany' function to be a function.");

	var self = this;

	return new self.Constructor({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), 
				function (pair) {
					var newValues = selector(pair[1]);
					if (!Object.isArray(newValues) &&
						!(newValues instanceof Series) &&
						!(newValues instanceof DataFrame)) {
						throw new Error("Expected return value from 'Series.selectMany' selector to be an array, a Series or a DataFrame, each item in the data sequence represents a new value in the resulting series.");
					}

					if (newValues instanceof DataFrame) {
						newValues = newValues.toValues();
					}
					else if (newValues instanceof Series)
					{
						newValues = newValues.toValues();
					}

					var newPairs = [];
					for (var newValueIndex = 0; newValueIndex < newValues.length; ++newValueIndex) {
						newPairs.push([pair[0], newValues[newValueIndex]]);
					}

					return newPairs;
				}
			);
		},
	}); 	
};

/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each value to a different data structure.
 */
Series.prototype.selectManyPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'Series.selectManyPairs' function to be a function.");

	var self = this;

	return new self.Constructor({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), 
				function (pair) {
					var newPairs = selector(pair[1], pair[0]);
					if (!Object.isArray(newPairs)) {
						throw new Error("Expected return value from 'Series.selectManyPairs' selector to be an array of pairs, each item in the array represents a new pair in the resulting series.");
					}

					for (var pairIndex = 0; pairIndex < newPairs.length; ++pairIndex) {
						var newPair = newPairs[pairIndex];
						if (!Object.isArray(newPair) || newPair.length !== 2) {
							throw new Error("Expected return value from 'Series.selectManyPairs' selector to be am array of pairs, but item at index " + pairIndex + " is not an array with two items: [index, value].");
						}
					}

					return newPairs;
				}
			);
		},
	}); 	
};

//
// Throw an exception if the sort method doesn't make sense.
//
var validateSortMethod = function (sortMethod) {
	assert.isString(sortMethod);
	assert(
		sortMethod === 'orderBy' || 
	   sortMethod === 'orderByDescending' ||
	   sortMethod === 'thenBy' ||
	   sortMethod === 'thenByDescending', 
	   "Expected 'sortMethod' to be one of 'orderBy', 'orderByDescending', 'thenBy' or 'thenByDescending', instead it is '" + sortMethod + "'."
   );
};

//
// Execute a batched sorting command.
//
var executeOrderBy = function (self, batch) {

	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);

	//
	// Don't invoke the sort until we really know what we need.
	//
	var executeLazySort = function () {

		batch.forEach(function (orderCmd) {
			assert.isObject(orderCmd);
			assert.isFunction(orderCmd.sortSelector);
			validateSortMethod(orderCmd.sortMethod);
		});

		var pairs = self.toPairs();

		return E.from(batch)
			.aggregate(E.from(pairs), function (unsorted, orderCmd) {
				return unsorted[orderCmd.sortMethod](function (pair) {
					return orderCmd.sortSelector(pair[1]);
				}); 
			})
			.toArray();
	};

	return new self.Constructor({
		iterable: function () {
			return new ArrayIterator(executeLazySort());
		},

		considerAllRows: true, //todo: This is currently needed to have df columns survive sorting. 
	});
};

//
// Order by values in a partcular series, either ascending or descending
//
var orderBy = function (self, sortMethod, sortSelector) {
	assert.isObject(self);
	validateSortMethod(sortMethod);
	assert.isFunction(sortSelector);

	var batchOrder = [
		{ 
			sortSelector: sortSelector, 
			sortMethod: sortMethod 
		}
	];

	var sortedDataFrame = executeOrderBy(self, batchOrder);
	sortedDataFrame.thenBy = orderThenBy(self, batchOrder, 'thenBy');
	sortedDataFrame.thenByDescending = orderThenBy(self, batchOrder, 'thenByDescending');	
	return sortedDataFrame;
};

//
// Generates a thenBy function that is attached to already ordered data frames.
//
var orderThenBy = function (self, batch, nextSortMethod) {
	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);
	validateSortMethod(nextSortMethod);
	
	return function (sortSelector) {
		assert.isFunction(sortSelector, "Expected parameter 'sortSelector' to be a function");

		var extendedBatch = batch.concat([
			{
				sortSelector: sortSelector,
				sortMethod: nextSortMethod,
			},
		]);

		var sortedDataFrame = executeOrderBy(self, extendedBatch);
		sortedDataFrame.thenBy = orderThenBy(self, extendedBatch, 'thenBy');
		sortedDataFrame.thenByDescending = orderThenBy(self, extendedBatch, 'thenByDescending');		
		return sortedDataFrame;
	};	
};

/**
 * Sorts the series by sort selector (ascending). 
 * 
 * @param {function} sortSelector - An function to select a value to sort by.
 */
Series.prototype.orderBy = function (sortSelector) {

	assert.isFunction

	var self = this;
	return orderBy(self, 'orderBy', sortSelector);
};

/**
 * Sorts the series by sort selector (descending). 
 * 
 * @param {function} sortSelector - An function to select a value to sort by.
 */
Series.prototype.orderByDescending = function (sortSelector) {

	var self = this;
	return orderBy(self, 'orderByDescending', sortSelector);
};

/**
 * Create a new series from a slice of rows.
 *
 * @param {int|function} startIndexOrStartPredicate - Index where the slice starts or a predicate function that determines where the slice starts.
 * @param {int|function} endIndexOrEndPredicate - Marks the end of the slice, one row past the last row to include. Or a predicate function that determines when the slice has ended.
 * @param {function} [predicate] - Optional predicate to compare index against start/end index. Return true to start or stop the slice.
 */
Series.prototype.slice = function (startIndexOrStartPredicate, endIndexOrEndPredicate, predicate) {

	var self = this;

	var startIndex;
	var endIndex;
	var startPredicate = null;
	var endPredicate = null;

	if (predicate) {
		assert.isFunction(predicate, "Expected 'predicate' parameter to slice function to be function.");
	}

	if (Object.isFunction(startIndexOrStartPredicate)) {
		startPredicate = startIndexOrStartPredicate;
	}
	else {
		startIndex = startIndexOrStartPredicate;
		startPredicate = function (value) {
				return predicate && predicate(value, startIndex) || value < startIndex;
			};
	}

	if (Object.isFunction(endIndexOrEndPredicate)) {
		endPredicate = endIndexOrEndPredicate;
	}
	else {
		endIndex = endIndexOrEndPredicate;
		endPredicate = function (value) {
				return predicate && predicate(value, endIndex) || value < endIndex;
			};
	}

	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new TakeWhileIterator(
					new SkipWhileIterator(
						self.getIterator(),
						function (pair) {
							return startPredicate(pair[0]); // Check index for start condition.
						}
					),
					function (pair) {
						return endPredicate(pair[0]); // Check index for end condition.
					}
				);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		},
	});
};

/**
 * Segment a Series into 'windows'. Returns a new Series. Each value in the new Series contains a 'window' (or segment) of the original Series.
 * Use select or selectPairs to aggregate.
 *
 * @param {integer} period - The number of values in the window.
 */
Series.prototype.window = function (period, obsoleteSelector) {

	assert.isNumber(period, "Expected 'period' parameter to 'window' to be a number.");
	assert(!obsoleteSelector, "Selector parameter is obsolete and no longer required.");

	var self = this;

	return new Series({
		iterable: function () {

			var curOutput = undefined;
			var windowIndex = 0;

			return {
				moveNext: function () {
					var window = self.skip(windowIndex*period).take(period);
					if (window.none(function () { return true; })) { //todo: Shouldn't have to pass a predicate.
						return false; // Nothing left.
					}

					curOutput = [
						windowIndex, 
						window,						
					];
					++windowIndex;
					return true;
				},

				getCurrent: function () {
					return curOutput;
				},
			};

		}
	});	
};

/** 
 * Segment a Series into 'rolling windows'. Returns a new Series. Each value in the new Series contains a 'window' (or segment) of the original Series.
 * Use select or selectPairs to aggregate.
 *
 * @param {integer} period - The number of values in the window.
 */
Series.prototype.rollingWindow = function (period, obsoleteSelector) {

	assert.isNumber(period, "Expected 'period' parameter to 'rollingWindow' to be a number.");
	assert(!obsoleteSelector, "Selector parameter is obsolete and no longer required.");

	var self = this;

	return new Series({
		iterable: function () {

			var curOutput = undefined;
			var done = false;
			var windowIndex = 0;

			return {
				moveNext: function () {
					var window = self.skip(windowIndex).take(period);
					if (window.count() < period) {
						return false;
					}

					curOutput = [
						windowIndex, 
						window,						
					];
					++windowIndex;
					return true;
				},

				getCurrent: function () {
					return curOutput;
				},
			};

		}
	});
};

/** 
 * Format the data frame for display as a string.
 */
Series.prototype.toString = function () {

	var self = this;
	var Table = require('easy-table');

	var index = self.getIndex().toValues();
	var header = ["__index__", "__value__"];
	var rows = E.from(self.toValues())
			.select(function (value, rowIndex) { 
				return [index[rowIndex], value];
			})
			.toArray()

	var t = new Table();
	rows.forEach(function (row, rowIndex) {
		row.forEach(function (cell, cellIndex) {
			t.cell(header[cellIndex], cell);
		});
		t.newRow();
	});

	return t.toString();
};

/**
 * Compute the percent change for each row after the first.
 * Percentages are expressed as 0-1 values.
 */
Series.prototype.percentChange = function () {

	var self = this;
	return self
		.rollingWindow(2)
		.selectPairs(function (window) {
			var values = window.toValues();
			var amountChange = values[1] - values[0]; // Compute amount of change.
			var pctChange = amountChange / values[0]; // Compute % change.
			return [window.getIndex().last(), pctChange]; // Return new index and value.
		});
};

/**
 * Parse a series with string values to a series with int values.
 */
Series.prototype.parseInts = function () {

	var self = this;
	return self.select(function (value, valueIndex) {
		if (value === undefined) {
			return undefined;
		}
		else {
			assert.isString(value, "Called parseInts on series, expected all values in the series to be strings, instead found a '" + typeof(value) + "' at index " + valueIndex);

			if (value.length === 0) {
				return undefined;
			}

			return parseInt(value);
		}
	});
};

/**
 * Parse a series with string values to a series with float values.
 */
Series.prototype.parseFloats = function () {

	var self = this;
	return self.select(function (value, valueIndex) {
		if (value === undefined) {
			return undefined;
		}
		else {
			assert.isString(value, "Called parseFloats on series, expected all values in the series to be strings, instead found a '" + typeof(value) + "' at index " + valueIndex);

			if (value.length === 0) {
				return undefined;
			}

			return parseFloat(value);
		}
	});
};

/**
 * Parse a series with string values to a series with date values.
 *
 * @param {string} [formatString] - Optional formatting string for dates.
 */
Series.prototype.parseDates = function (formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	return self.select(function (value, valueIndex) {
		if (value === undefined) {
			return undefined;
		}
		else {
			assert.isString(value, "Called parseDates on series, expected all values in the series to be strings, instead found a '" + typeof(value) + "' at index " + valueIndex);

			if (value.length === 0) {
				return undefined;
			}

			return moment(value, formatString).toDate();
		}
	});
};

/**
 * Convert a series of values of different types to a series of string values.
 *
 * @param {string} [formatString] - Optional formatting string for dates.
 */
Series.prototype.toStrings = function (formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	return self.select(function (value) {
		if (value === undefined) {
			return undefined;
		}
		else if (value === null) {
			return null;
		}
		else if (formatString && Object.isDate(value)) {
			return moment(value).format(formatString);
		}
		else if (formatString && moment.isMoment(value)) {
			return value.format(formatString);
		}
		else {
			return value.toString();	
		}		
	});
};

/** 
  * Detect the actual types of the values that comprised the series and their frequency.
  * Returns a new series containing the type information.
  */
Series.prototype.detectTypes = function () {

	var self = this;

	return new DataFrame({
		columnNames: ["Type", "Frequency"],
		values: function () { //todo: make this properly lazy.
			var values = self.toValues();
			var totalValues = values.length;

			var typeFrequencies = E.from(values)
				.select(function (value) {
					var valueType = typeof(value);
					if (valueType === 'object') {
						if (Object.isDate(value)) {
							valueType = 'date';
						}
					}
					return valueType;
				})
				.aggregate({}, function (accumulated, valueType) {
					var typeInfo = accumulated[valueType];
					if (!typeInfo) {
						typeInfo = {
							count: 0
						};
						accumulated[valueType] = typeInfo;
					}
					++typeInfo.count;
					return accumulated;
				});

			return new ArrayIterator(
				E.from(Object.keys(typeFrequencies))
					.select(function (valueType) {
						return [
							valueType,
							(typeFrequencies[valueType].count / totalValues) * 100
						];
					})
					.toArray()
			);
		},		
	});
};

/** 
  * Detect the frequency of values in the series.
  * Returns a new series containing the information.
  */
Series.prototype.detectValues = function () {

	var self = this;

	return new DataFrame({
		columnNames: ["Value", "Frequency"],
		values: function () {
			var values = self.toValues();
			var totalValues = values.length;

			var valueFrequencies = E.from(values)
				.aggregate({}, function (accumulated, value) {
					var valueKey = value.toString() + "-" + typeof(value);
					var valueInfo = accumulated[valueKey];
					if (!valueInfo) {
						valueInfo = {
							count: 0,
							value: value,
						};
						accumulated[valueKey] = valueInfo;
					}
					++valueInfo.count;
					return accumulated;
				});

			return new ArrayIterator(
				E.from(Object.keys(valueFrequencies))
					.select(function (valueKey) {
						var valueInfo = valueFrequencies[valueKey];
						return [
							valueInfo.value,
							(valueInfo.count / totalValues) * 100
						];
					})
					.toArray()
			);
		},
	});
};

/**
 * Produces a new series with all string values truncated to the requested maximum length.
 *
 * @param {int} maxLength - The maximum length of the string values after truncation.
 */
Series.prototype.truncateStrings = function (maxLength) {
	assert.isNumber(maxLength, "Expected 'maxLength' parameter to 'truncateStrings' to be an integer.");

	var self = this;
	return self
		.select(function (value) {
			if (Object.isString(value)) {
				if (value.length > maxLength) {
					return value.substring(0, maxLength);
				}
			}

			return value;
		});
};

/*
 * Extract values from the series. This forces lazy evaluation to complete.
 */
Series.prototype.toValues = function () {

	var self = this;
	var iterator = self.getIterator();
	validateIterator(iterator);

	var values = [];

	while (iterator.moveNext()) {
		var value = iterator.getCurrent()[1]; // Extract value.
		if (value !== undefined) {
			values.push(value);
		}
	}

	return values;
};

/**
 * Forces lazy evaluation to complete and 'bakes' the series into memory.
 */
Series.prototype.bake = function () {

	var self = this;
	if (self._baked) {
		// Already baked, just return self.
		return self;
	}

	var pairs = self.toPairs();
	var baked = new Series({
		iterable: function () {
			return new ArrayIterator(pairs);
		},
	});
	baked._baked = true;
	return baked;
};

/**
 * Retreive the data as pairs of [index, value].
 */
Series.prototype.toPairs = function () {

	var self = this;
	var iterator = self.getIterator();
	validateIterator(iterator);

	var pairs = [];

	while (iterator.moveNext()) {
		var pair = iterator.getCurrent();
		if (pair[1] !== undefined) {
			pairs.push(pair);
		}		
	}

	return pairs;
};

/**
 * Count the number of rows in the series.
 */
Series.prototype.count = function () {

	var self = this;
	var total = 0;
	var iterator = self.getIterator();

	while (iterator.moveNext()) {
		++total;
	}

	return total;
};

/**
 * Get the first value of the Series.
 */
Series.prototype.first = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	return iterator.getCurrent()[1]; // Extract value.
};

/**
 * Get the last value of the Series.
 */
Series.prototype.last = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	while (iterator.moveNext()) {
		; // Don't evaluate each item, it's too expensive.
	}

	return iterator.getCurrent()[1]; // Just evaluate the last item of the iterator.
};

/**
 * Get the first index/value pair of the Series.
 */
Series.prototype.firstPair = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	return iterator.getCurrent();
};

/**
 * Get the last index/value pair of the Series.
 */
Series.prototype.lastPair = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	while (iterator.moveNext()) {
		; // Don't evaluate each item, it's too expensive.
	}

	return iterator.getCurrent();
};

/**
 * Get the first index of the Series.
 */
Series.prototype.firstIndex = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	return iterator.getCurrent()[0]; // Extract index.
};

/**
 * Get the last index of the Series.
 */
Series.prototype.lastIndex = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	while (iterator.moveNext()) {
		; // Don't evaluate each item, it's too expensive.
	}

	return iterator.getCurrent()[0]; // Extract index.
};

/** 
 * Reverse the series.
 */
Series.prototype.reverse = function () {

	var self = this;
	return new self.Constructor({
		iterable: function () {
			var pairs = [];
			var iterator = self.getIterator();

			while (iterator.moveNext()) {
				pairs.push(iterator.getCurrent());
			}

			return new ArrayIterator(pairs.reverse());
		},
	});
};

/** 
 * Inflate a series to a data-frame.
 *
 * @param {function} [selector] - Optional selector function that transforms each value in the series to a row in the new data-frame.
 */
Series.prototype.inflate = function (selector) {

	var self = this;

	if (selector) {
		assert.isFunction(selector, "Expected 'selector' parameter to 'inflate' function to be a function.");
	}
	else {
		selector = function (value) {
			return value;
		}
	}

	return new DataFrame({
		__iterable: new SelectIterable(self, selector),<commit_msg>Moved more fns over to '__iterable'.<commit_after>'use strict';

// 
// Base class for series classes.
//

var assert = require('chai').assert; 
var E = require('linq');
var moment = require('moment');
var ArrayIterator = require('./iterators/array');
var validateIterator = require('./iterators/validate');
var SkipIterator = require('./iterators/skip');
var SkipWhileIterator = require('./iterators/skip-while');
var TakeIterator = require('../src/iterators/take');
var TakeWhileIterator = require('../src/iterators/take-while');
var SelectIterator = require('../src/iterators/select');
var SelectManyIterator = require('../src/iterators/select-many');
var PairIterator = require('../src/iterators/pair');
var WhereIterator = require('../src/iterators/where');
var CountIterator = require('../src/iterators/count');
var EmptyIterator = require('../src/iterators/empty');
var PairsIterable = require('../src/iterables/pairs');
var SelectIterable = require('../src/iterables/select');
var ArrayIterable = require('../src/iterables/array');
var extend = require('extend');

//
// Represents a time series.
//
var Series = function (config) {

	var self = this;

	if (!self.Constructor) {
		self.Constructor = Series;
	}

	if (!config) {
		self.__iterable = {
			getIterator: function () {
				return new EmptyIterator();
			},
		};
		return;
	}

	assert.isObject(config, "Expected 'config' parameter to Series constructor to be an object with options for initialisation.");

	if (config.__iterable) {
		// Setting the inner iterable directly.
		// Power to you!
		self.__iterable = config.__iterable;
		return;
	}

	if (config.iterable) {

		if (Object.isFunction(config.iterable)) {
			self.__iterable = {
				getIterator: config.iterable
			};
		}
		else {
			assert.isObject(config.iterable, "Expected 'iterable' field of 'config' parameter to Series constructor to either be a function that returns a pairs iterator or an object with a getIterator function.");
			assert.isFunction(config.iterable.getIterator, "Expected 'iterable' field of 'config' parameter to Series constructor to either be a function that returns a pairs iterator or an object with a getIterator function.");

			self.__iterable = config.iterable; 
		}
		return;
	}

	if (config.values) {
		if (!Object.isFunction(config.values)) {
			assert.isArray(config.values, "Expected 'values' field of 'config' parameter to Series constructor be an array of values or a function that returns an iterator.");
		}
	}

	if (config.index) {
		if (!Object.isFunction(config.index) && !Object.isArray(config.index)) {
			assert.isObject(config.index, "Expected 'index' field of 'config' parameter to Series constructor to be an array, function or Series object.");
		}
	}

	var indexIterable = null;
	var valuesIterable = null;

	var index = config.index;
	if (!index) {
		indexIterable = {
			getIterator: function () {
				return new CountIterator();
			},			
		};
	}
	else if (Object.isFunction(index)) {
		indexIterable = {
			getIterator: index,
		};
	}
	else if (Object.isArray(index)) {
		indexIterable = {
			getIterator: function () {
				return new ArrayIterator(index);
			},
		};
	}
	else {		
		indexIterable = {
			getIterator: function () {
				return new SelectIterator(
					index.getIterator(),
					function (pair) {
						return pair[1]; // Extract the value from the index.
					}
				)

			},
		};
	}

	var values = config.values;
	if (!values) {
		valuesIterable = {
			getIterator: function () {
				return new EmptyIterator();
			},			
		};
	}
	else if (Object.isFunction(values)) {
		valuesIterable = {
			getIterator: values,
		};
	}
	else {
		valuesIterable = {
			getIterator: function () {
				return new ArrayIterator(values);
			},
		};
	}

	self.__iterable = new PairsIterable(indexIterable, valuesIterable);
};

module.exports = Series;

var concatSeries = require('./concat-series');
var DataFrame = require('./dataframe');
var zipSeries = require('./zip-series');

/**
 * Get an iterator for index & values of the series.
 */
Series.prototype.getIterator = function () {
	var self = this;
	return self.__iterable.getIterator();
};

/**
 * Retreive the index of the series.
 */
Series.prototype.getIndex = function () {
	var self = this;
	return new Series({
		values: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					return pair[0]; // Extract the index.
				}
			);
		},
	});
};

/**
 * Apply a new index to the Series.
 * 
 * @param {array|Series} newIndex - The new index to apply to the Series.
 */
Series.prototype.withIndex = function (newIndex) {

	if (!Object.isArray(newIndex)) {
		assert.isObject(newIndex, "'Expected 'newIndex' parameter to 'Series.withIndex' to be an array or Series object.");
	}	

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				var indexIterator = Object.isArray(newIndex)
					? new ArrayIterator(newIndex)
					: new SelectIterator(
						newIndex.getIterator(),
						function (pair) {
							return pair[1]; // Extract index value.
						}
					)
					;
				return new PairIterator(
					indexIterator,
					new SelectIterator(
						self.getIterator(),
						function (pair) {
							return pair[1]; // Extract value.
						}
					)
				);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		}
	});
};

/**
 * Reset the index of the data frame back to the default sequential integer index.
 */
Series.prototype.resetIndex = function () {

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new PairIterator(
					new CountIterator(), // Reset index.
					new SelectIterator(
						self.getIterator(),
						function (pair) {
							return pair[1]; // Extract value.
						}
					)
				)
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		},
	});
};

/**
 * Skip a number of rows in the series.
 *
 * @param {int} numRows - Number of rows to skip.
 */
Series.prototype.skip = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'skip' function to be a number.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new SkipIterator(self.getIterator(), numRows);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		}
	}); 	
};

/**
 * Skips values in the series while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.skipWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new self.Constructor({
		__iterable: {

			getIterator: function () {
				return new SkipWhileIterator(
					self.getIterator(),
					function (pair) {
						return predicate(pair[1]); // Extract the value.
					}
				); 
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		}
	}); 	
};

/**
 * Skips values in the series until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.skipUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.skipWhile(function (value) { 
		return !predicate(value); 
	});
};

/**
 * Take a number of rows in the series.
 *
 * @param {int} numRows - Number of rows to take.
 */
Series.prototype.take = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'take' function to be a number.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new TakeIterator(self.getIterator(), numRows);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		},
	});
};

/**
 * Take values from the series while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.takeWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new TakeWhileIterator(
					self.getIterator(),
					function (pair) {
						return predicate(pair[1]); // Extract the value.
					}
				); 
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},

		},
	}); 	
};

/**
 * Take values from the series until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
Series.prototype.takeUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.takeWhile(function (value) { 
		return !predicate(value); 
	});
};

/**
 * Filter a series by a predicate selector.
 *
 * @param {function} filterSelectorPredicate - Predicte function to filter rows of the series.
 */
Series.prototype.where = function (filterSelectorPredicate) {
	assert.isFunction(filterSelectorPredicate, "Expected 'filterSelectorPredicate' parameter to 'where' function to be a function.");

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new WhereIterator(
					self.getIterator(), 
					function (pair) {
						return filterSelectorPredicate(pair[1]); // Extract the value.
					}
				);
			},

			getColumnNames: function () {
				return self.getColumnNames(); // Doesn't change column names, or order.
			}
		},
	}); 	
};


/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each value to create a new series.
 */
Series.prototype.select = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'select' function to be a function.");

	var self = this;
	return new self.Constructor({
		__iterable: new SelectIterable(self, selector),
	}); 	
};

/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each index/value to a create a new series.
 */
Series.prototype.selectPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'selectPairs' function to be a function.");

	var self = this;
	return new self.Constructor({
		iterable: function () {
			return new SelectIterator(self.getIterator(), 
				function (pair) {
					var newPair = selector(pair[1], pair[0]);
					if (!Object.isArray(newPair) || newPair.length !== 2) {
						throw new Error("Expected return value from 'Series.selectPairs' selector to be a pair, that is an array with two items: [index, value].");
					}
					return newPair;
				}
			);
		},		
	}); 	
};

/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each value to a different data structure.
 */
Series.prototype.selectMany = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'Series.selectMany' function to be a function.");

	var self = this;

	return new self.Constructor({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), 
				function (pair) {
					var newValues = selector(pair[1]);
					if (!Object.isArray(newValues) &&
						!(newValues instanceof Series) &&
						!(newValues instanceof DataFrame)) {
						throw new Error("Expected return value from 'Series.selectMany' selector to be an array, a Series or a DataFrame, each item in the data sequence represents a new value in the resulting series.");
					}

					if (newValues instanceof DataFrame) {
						newValues = newValues.toValues();
					}
					else if (newValues instanceof Series)
					{
						newValues = newValues.toValues();
					}

					var newPairs = [];
					for (var newValueIndex = 0; newValueIndex < newValues.length; ++newValueIndex) {
						newPairs.push([pair[0], newValues[newValueIndex]]);
					}

					return newPairs;
				}
			);
		},
	}); 	
};

/**
 * Generate a new series based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each value to a different data structure.
 */
Series.prototype.selectManyPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'Series.selectManyPairs' function to be a function.");

	var self = this;

	return new self.Constructor({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), 
				function (pair) {
					var newPairs = selector(pair[1], pair[0]);
					if (!Object.isArray(newPairs)) {
						throw new Error("Expected return value from 'Series.selectManyPairs' selector to be an array of pairs, each item in the array represents a new pair in the resulting series.");
					}

					for (var pairIndex = 0; pairIndex < newPairs.length; ++pairIndex) {
						var newPair = newPairs[pairIndex];
						if (!Object.isArray(newPair) || newPair.length !== 2) {
							throw new Error("Expected return value from 'Series.selectManyPairs' selector to be am array of pairs, but item at index " + pairIndex + " is not an array with two items: [index, value].");
						}
					}

					return newPairs;
				}
			);
		},
	}); 	
};

//
// Throw an exception if the sort method doesn't make sense.
//
var validateSortMethod = function (sortMethod) {
	assert.isString(sortMethod);
	assert(
		sortMethod === 'orderBy' || 
	   sortMethod === 'orderByDescending' ||
	   sortMethod === 'thenBy' ||
	   sortMethod === 'thenByDescending', 
	   "Expected 'sortMethod' to be one of 'orderBy', 'orderByDescending', 'thenBy' or 'thenByDescending', instead it is '" + sortMethod + "'."
   );
};

//
// Execute a batched sorting command.
//
var executeOrderBy = function (self, batch) {

	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);

	//
	// Don't invoke the sort until we really know what we need.
	//
	var executeLazySort = function () {

		batch.forEach(function (orderCmd) {
			assert.isObject(orderCmd);
			assert.isFunction(orderCmd.sortSelector);
			validateSortMethod(orderCmd.sortMethod);
		});

		var pairs = self.toPairs();

		return E.from(batch)
			.aggregate(E.from(pairs), function (unsorted, orderCmd) {
				return unsorted[orderCmd.sortMethod](function (pair) {
					return orderCmd.sortSelector(pair[1]);
				}); 
			})
			.toArray();
	};

	return new self.Constructor({
		iterable: function () {
			return new ArrayIterator(executeLazySort());
		},

		considerAllRows: true, //todo: This is currently needed to have df columns survive sorting. 
	});
};

//
// Order by values in a partcular series, either ascending or descending
//
var orderBy = function (self, sortMethod, sortSelector) {
	assert.isObject(self);
	validateSortMethod(sortMethod);
	assert.isFunction(sortSelector);

	var batchOrder = [
		{ 
			sortSelector: sortSelector, 
			sortMethod: sortMethod 
		}
	];

	var sortedDataFrame = executeOrderBy(self, batchOrder);
	sortedDataFrame.thenBy = orderThenBy(self, batchOrder, 'thenBy');
	sortedDataFrame.thenByDescending = orderThenBy(self, batchOrder, 'thenByDescending');	
	return sortedDataFrame;
};

//
// Generates a thenBy function that is attached to already ordered data frames.
//
var orderThenBy = function (self, batch, nextSortMethod) {
	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);
	validateSortMethod(nextSortMethod);
	
	return function (sortSelector) {
		assert.isFunction(sortSelector, "Expected parameter 'sortSelector' to be a function");

		var extendedBatch = batch.concat([
			{
				sortSelector: sortSelector,
				sortMethod: nextSortMethod,
			},
		]);

		var sortedDataFrame = executeOrderBy(self, extendedBatch);
		sortedDataFrame.thenBy = orderThenBy(self, extendedBatch, 'thenBy');
		sortedDataFrame.thenByDescending = orderThenBy(self, extendedBatch, 'thenByDescending');		
		return sortedDataFrame;
	};	
};

/**
 * Sorts the series by sort selector (ascending). 
 * 
 * @param {function} sortSelector - An function to select a value to sort by.
 */
Series.prototype.orderBy = function (sortSelector) {

	assert.isFunction

	var self = this;
	return orderBy(self, 'orderBy', sortSelector);
};

/**
 * Sorts the series by sort selector (descending). 
 * 
 * @param {function} sortSelector - An function to select a value to sort by.
 */
Series.prototype.orderByDescending = function (sortSelector) {

	var self = this;
	return orderBy(self, 'orderByDescending', sortSelector);
};

/**
 * Create a new series from a slice of rows.
 *
 * @param {int|function} startIndexOrStartPredicate - Index where the slice starts or a predicate function that determines where the slice starts.
 * @param {int|function} endIndexOrEndPredicate - Marks the end of the slice, one row past the last row to include. Or a predicate function that determines when the slice has ended.
 * @param {function} [predicate] - Optional predicate to compare index against start/end index. Return true to start or stop the slice.
 */
Series.prototype.slice = function (startIndexOrStartPredicate, endIndexOrEndPredicate, predicate) {

	var self = this;

	var startIndex;
	var endIndex;
	var startPredicate = null;
	var endPredicate = null;

	if (predicate) {
		assert.isFunction(predicate, "Expected 'predicate' parameter to slice function to be function.");
	}

	if (Object.isFunction(startIndexOrStartPredicate)) {
		startPredicate = startIndexOrStartPredicate;
	}
	else {
		startIndex = startIndexOrStartPredicate;
		startPredicate = function (value) {
				return predicate && predicate(value, startIndex) || value < startIndex;
			};
	}

	if (Object.isFunction(endIndexOrEndPredicate)) {
		endPredicate = endIndexOrEndPredicate;
	}
	else {
		endIndex = endIndexOrEndPredicate;
		endPredicate = function (value) {
				return predicate && predicate(value, endIndex) || value < endIndex;
			};
	}

	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				return new TakeWhileIterator(
					new SkipWhileIterator(
						self.getIterator(),
						function (pair) {
							return startPredicate(pair[0]); // Check index for start condition.
						}
					),
					function (pair) {
						return endPredicate(pair[0]); // Check index for end condition.
					}
				);
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		},
	});
};

/**
 * Segment a Series into 'windows'. Returns a new Series. Each value in the new Series contains a 'window' (or segment) of the original Series.
 * Use select or selectPairs to aggregate.
 *
 * @param {integer} period - The number of values in the window.
 */
Series.prototype.window = function (period, obsoleteSelector) {

	assert.isNumber(period, "Expected 'period' parameter to 'window' to be a number.");
	assert(!obsoleteSelector, "Selector parameter is obsolete and no longer required.");

	var self = this;

	return new Series({
		iterable: function () {

			var curOutput = undefined;
			var windowIndex = 0;

			return {
				moveNext: function () {
					var window = self.skip(windowIndex*period).take(period);
					if (window.none(function () { return true; })) { //todo: Shouldn't have to pass a predicate.
						return false; // Nothing left.
					}

					curOutput = [
						windowIndex, 
						window,						
					];
					++windowIndex;
					return true;
				},

				getCurrent: function () {
					return curOutput;
				},
			};

		}
	});	
};

/** 
 * Segment a Series into 'rolling windows'. Returns a new Series. Each value in the new Series contains a 'window' (or segment) of the original Series.
 * Use select or selectPairs to aggregate.
 *
 * @param {integer} period - The number of values in the window.
 */
Series.prototype.rollingWindow = function (period, obsoleteSelector) {

	assert.isNumber(period, "Expected 'period' parameter to 'rollingWindow' to be a number.");
	assert(!obsoleteSelector, "Selector parameter is obsolete and no longer required.");

	var self = this;

	return new Series({
		iterable: function () {

			var curOutput = undefined;
			var done = false;
			var windowIndex = 0;

			return {
				moveNext: function () {
					var window = self.skip(windowIndex).take(period);
					if (window.count() < period) {
						return false;
					}

					curOutput = [
						windowIndex, 
						window,						
					];
					++windowIndex;
					return true;
				},

				getCurrent: function () {
					return curOutput;
				},
			};

		}
	});
};

/** 
 * Format the data frame for display as a string.
 */
Series.prototype.toString = function () {

	var self = this;
	var Table = require('easy-table');

	var index = self.getIndex().toValues();
	var header = ["__index__", "__value__"];
	var rows = E.from(self.toValues())
			.select(function (value, rowIndex) { 
				return [index[rowIndex], value];
			})
			.toArray()

	var t = new Table();
	rows.forEach(function (row, rowIndex) {
		row.forEach(function (cell, cellIndex) {
			t.cell(header[cellIndex], cell);
		});
		t.newRow();
	});

	return t.toString();
};

/**
 * Compute the percent change for each row after the first.
 * Percentages are expressed as 0-1 values.
 */
Series.prototype.percentChange = function () {

	var self = this;
	return self
		.rollingWindow(2)
		.selectPairs(function (window) {
			var values = window.toValues();
			var amountChange = values[1] - values[0]; // Compute amount of change.
			var pctChange = amountChange / values[0]; // Compute % change.
			return [window.getIndex().last(), pctChange]; // Return new index and value.
		});
};

/**
 * Parse a series with string values to a series with int values.
 */
Series.prototype.parseInts = function () {

	var self = this;
	return self.select(function (value, valueIndex) {
		if (value === undefined) {
			return undefined;
		}
		else {
			assert.isString(value, "Called parseInts on series, expected all values in the series to be strings, instead found a '" + typeof(value) + "' at index " + valueIndex);

			if (value.length === 0) {
				return undefined;
			}

			return parseInt(value);
		}
	});
};

/**
 * Parse a series with string values to a series with float values.
 */
Series.prototype.parseFloats = function () {

	var self = this;
	return self.select(function (value, valueIndex) {
		if (value === undefined) {
			return undefined;
		}
		else {
			assert.isString(value, "Called parseFloats on series, expected all values in the series to be strings, instead found a '" + typeof(value) + "' at index " + valueIndex);

			if (value.length === 0) {
				return undefined;
			}

			return parseFloat(value);
		}
	});
};

/**
 * Parse a series with string values to a series with date values.
 *
 * @param {string} [formatString] - Optional formatting string for dates.
 */
Series.prototype.parseDates = function (formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	return self.select(function (value, valueIndex) {
		if (value === undefined) {
			return undefined;
		}
		else {
			assert.isString(value, "Called parseDates on series, expected all values in the series to be strings, instead found a '" + typeof(value) + "' at index " + valueIndex);

			if (value.length === 0) {
				return undefined;
			}

			return moment(value, formatString).toDate();
		}
	});
};

/**
 * Convert a series of values of different types to a series of string values.
 *
 * @param {string} [formatString] - Optional formatting string for dates.
 */
Series.prototype.toStrings = function (formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	return self.select(function (value) {
		if (value === undefined) {
			return undefined;
		}
		else if (value === null) {
			return null;
		}
		else if (formatString && Object.isDate(value)) {
			return moment(value).format(formatString);
		}
		else if (formatString && moment.isMoment(value)) {
			return value.format(formatString);
		}
		else {
			return value.toString();	
		}		
	});
};

/** 
  * Detect the actual types of the values that comprised the series and their frequency.
  * Returns a new series containing the type information.
  */
Series.prototype.detectTypes = function () {

	var self = this;

	return new DataFrame({
		columnNames: ["Type", "Frequency"],
		values: function () { //todo: make this properly lazy.
			var values = self.toValues();
			var totalValues = values.length;

			var typeFrequencies = E.from(values)
				.select(function (value) {
					var valueType = typeof(value);
					if (valueType === 'object') {
						if (Object.isDate(value)) {
							valueType = 'date';
						}
					}
					return valueType;
				})
				.aggregate({}, function (accumulated, valueType) {
					var typeInfo = accumulated[valueType];
					if (!typeInfo) {
						typeInfo = {
							count: 0
						};
						accumulated[valueType] = typeInfo;
					}
					++typeInfo.count;
					return accumulated;
				});

			return new ArrayIterator(
				E.from(Object.keys(typeFrequencies))
					.select(function (valueType) {
						return [
							valueType,
							(typeFrequencies[valueType].count / totalValues) * 100
						];
					})
					.toArray()
			);
		},		
	});
};

/** 
  * Detect the frequency of values in the series.
  * Returns a new series containing the information.
  */
Series.prototype.detectValues = function () {

	var self = this;

	return new DataFrame({
		columnNames: ["Value", "Frequency"],
		values: function () {
			var values = self.toValues();
			var totalValues = values.length;

			var valueFrequencies = E.from(values)
				.aggregate({}, function (accumulated, value) {
					var valueKey = value.toString() + "-" + typeof(value);
					var valueInfo = accumulated[valueKey];
					if (!valueInfo) {
						valueInfo = {
							count: 0,
							value: value,
						};
						accumulated[valueKey] = valueInfo;
					}
					++valueInfo.count;
					return accumulated;
				});

			return new ArrayIterator(
				E.from(Object.keys(valueFrequencies))
					.select(function (valueKey) {
						var valueInfo = valueFrequencies[valueKey];
						return [
							valueInfo.value,
							(valueInfo.count / totalValues) * 100
						];
					})
					.toArray()
			);
		},
	});
};

/**
 * Produces a new series with all string values truncated to the requested maximum length.
 *
 * @param {int} maxLength - The maximum length of the string values after truncation.
 */
Series.prototype.truncateStrings = function (maxLength) {
	assert.isNumber(maxLength, "Expected 'maxLength' parameter to 'truncateStrings' to be an integer.");

	var self = this;
	return self
		.select(function (value) {
			if (Object.isString(value)) {
				if (value.length > maxLength) {
					return value.substring(0, maxLength);
				}
			}

			return value;
		});
};

/*
 * Extract values from the series. This forces lazy evaluation to complete.
 */
Series.prototype.toValues = function () {

	var self = this;
	var iterator = self.getIterator();
	validateIterator(iterator);

	var values = [];

	while (iterator.moveNext()) {
		var value = iterator.getCurrent()[1]; // Extract value.
		if (value !== undefined) {
			values.push(value);
		}
	}

	return values;
};

/**
 * Forces lazy evaluation to complete and 'bakes' the series into memory.
 */
Series.prototype.bake = function () {

	var self = this;
	if (self._baked) {
		// Already baked, just return self.
		return self;
	}

	var pairs = self.toPairs();
	var baked = new Series({
		iterable: function () {
			return new ArrayIterator(pairs);
		},
	});
	baked._baked = true;
	return baked;
};

/**
 * Retreive the data as pairs of [index, value].
 */
Series.prototype.toPairs = function () {

	var self = this;
	var iterator = self.getIterator();
	validateIterator(iterator);

	var pairs = [];

	while (iterator.moveNext()) {
		var pair = iterator.getCurrent();
		if (pair[1] !== undefined) {
			pairs.push(pair);
		}		
	}

	return pairs;
};

/**
 * Count the number of rows in the series.
 */
Series.prototype.count = function () {

	var self = this;
	var total = 0;
	var iterator = self.getIterator();

	while (iterator.moveNext()) {
		++total;
	}

	return total;
};

/**
 * Get the first value of the Series.
 */
Series.prototype.first = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	return iterator.getCurrent()[1]; // Extract value.
};

/**
 * Get the last value of the Series.
 */
Series.prototype.last = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	while (iterator.moveNext()) {
		; // Don't evaluate each item, it's too expensive.
	}

	return iterator.getCurrent()[1]; // Just evaluate the last item of the iterator.
};

/**
 * Get the first index/value pair of the Series.
 */
Series.prototype.firstPair = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	return iterator.getCurrent();
};

/**
 * Get the last index/value pair of the Series.
 */
Series.prototype.lastPair = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	while (iterator.moveNext()) {
		; // Don't evaluate each item, it's too expensive.
	}

	return iterator.getCurrent();
};

/**
 * Get the first index of the Series.
 */
Series.prototype.firstIndex = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	return iterator.getCurrent()[0]; // Extract index.
};

/**
 * Get the last index of the Series.
 */
Series.prototype.lastIndex = function () {

	var self = this;
	var iterator = self.getIterator();

	if (!iterator.moveNext()) {
		throw new Error("No values in Series.");
	}

	while (iterator.moveNext()) {
		; // Don't evaluate each item, it's too expensive.
	}

	return iterator.getCurrent()[0]; // Extract index.
};

/** 
 * Reverse the series.
 */
Series.prototype.reverse = function () {

	var self = this;
	return new self.Constructor({
		__iterable: {
			getIterator: function () {
				var pairs = [];
				var iterator = self.getIterator();

				while (iterator.moveNext()) {
					pairs.push(iterator.getCurrent());
				}

				return new ArrayIterator(pairs.reverse());
			},

			getColumnNames: function () {
				return self.getColumnNames();
			},
		}
	});
};

/** 
 * Inflate a series to a data-frame.
 *
 * @param {function} [selector] - Optional selector function that transforms each value in the series to a row in the new data-frame.
 */
Series.prototype.inflate = function (selector) {

	var self = this;

	if (selector) {
		assert.isFunction(selector, "Expected 'selector' parameter to 'inflate' function to be a function.");
	}
	else {
		selector = function (value) {
			return value;
		}
	}

	return new DataFrame({
		__iterable: new SelectIterable(self, selector),
{'fraction_non_alphanumeric': 0.12749523204068658, 'fraction_numerical': 0.001589319771137953, 'mean_word_length': 3.3402772987514657, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1945049', 'n_tokens_mistral': 21667, 'n_tokens_neox': 18480, 'n_words': 5757}