<?php

class it_pipe implements Iterator
{
	static $lastargfunc = array('explode' => 1, 'preg_match' => 1, 'preg_split' => 1, 'it__match' => 1, 'it__replace' => 1);

	var $lines; # all lines currently in pipe
	var $_valid;

/**
 * Creates a pipe object from input. Named arguments:
 * $p['fn'] filename to read lines from
 * $p['cmd'] cmd to read lines from
 * $p['data'] data to read line from. either array of lines without newslines or string
 */
function __construct($p = array())
{
	if ($p['data'])
		$this->lines = is_array($p['data']) ? $p['data'] : explode("\n", rtrim($p['data'], "\n"));
	else if ($p['cmd'])
		$this->lines = strlen($data = it::exec($p['cmd'], $p['args'])) ? explode("\n", rtrim($data, "\n")) : array();
	else
		foreach ((array)($p['fn'] ? $p['fn'] : "php://stdin") as $fn)
			$this->lines = array_merge((array)$this->lines, ($t = file($fn, FILE_IGNORE_NEW_LINES)) === false ? array() : $t); 
}

/**
 * Apply any php function to every line in pipe. Pass line as first arg except with some functions (e.g. explode)
 */
function __call($name, $params)
{
	list($parampos) = self::$lastargfunc[$name] ? array(count($params)) : array(0, array_unshift($params, ""));
	$func = ($t = it::match('(\w+)__(\w+)', $name)) ? array($t[0], $t[1]) : (method_exists($this, $name) ? array($this, $name) : $name);

	foreach($this->lines as $i => $params[$parampos])
		$this->lines[$i] = call_user_func_array($func, $params);

	return $this;
}

function __toString()
{
	return is_array($this->lines) ? join("\n", $this->lines) . "\n" : "";
}


#
# Internal: Implement iterator
#
function rewind()
{
	reset($this->lines);
	$this->next();
}

function next()
{
	$this->_valid = each($this->lines);
}

function valid()
{
	return $this->_valid;
}

function current()
{
	return $this->_valid[1];
}

function key()
{
	return $this->_valid[0];
}


/**
 * Apply an expression to every line
 */
function map($expr)
{
	$this->lines = it::map($expr, $this->lines);

	return $this;
}

/**
 * Convert pipe from utf8 to iso-latin
 */
function latin()
{
	return $this->map('utf8_decode($v)');
}

/**
 * Convert pipe from iso-latin to uft8
 */
function utf8()
{
	return $this->map('utf8_encode($v)');
}

/**
 * Return contents of pipe as key->val pair (key must be tab separated)
 */
function keyval()
{
	foreach ($this->lines as $line)
	{
		list($key, $val) = explode("\t", $line, 2);
		$result[$key] = $val;
	}

	return (array)$result;
}

/**
 * Swap first two columns
 */
function swap()
{
	foreach ($this->lines as $idx => $line)
	{
		list($col1, $col2) = explode("\t", $line, 2);
		$this->lines[$idx] = "$col2\t$col1";
	}

	return $this;
}

function ED()
{
	ED($this->lines);

	return $this;
}

/**
 * Convert every line into an object with named columns
 * @param collist comma separator column name list
 * @param separator split character ["\t"]
 */
function cols($collist, $separator = "\t")
{
	$keys = explode(",", $collist);
	foreach ($this->lines as $idx => $line)
	{
		if (count($keys) == count($arr = explode($separator, $line)))
			$this->lines[$idx] = (object)array_combine($keys, $arr);
		else
		{
			unset($this->lines[$idx]);
			foreach ($keys as $i => $key)
				$this->lines[$idx]->$key = $arr[$i];
		}
	}

	return $this;
}

/**
 * Return contents of pipe as associative records. Column titles read from first line, separators can be \t or ; or ,
 * @param $forceschema ignore schema in file, replace it with this comma separated column list
 */
function csv($forceschema = null)
{
	$counts = count_chars($this->lines[0]);
	$splitchar = $counts[ord("\t")] ? "\t" : ($counts[ord(";")] > $counts[ord(",")] ? ";" : ",");
	$csvhead = array_shift($this->lines);
	$cols = $forceschema ? explode(",", $forceschema) : str_getcsv(trim($csvhead, "#\n "), $splitchar, '"'); # should function_exists('str_getcsv')
	foreach (preg_grep('/^$/', $cols) as $idx => $dummy)
		$cols[$idx] = "field$idx"; # replace empty column names

	$oldlocale = setlocale(LC_CTYPE, "0");
	setlocale(LC_CTYPE, 'de_CH.iso-8859-1'); # this works for utf-8 as well

	foreach ($this->lines as $line)
		$records[] = (object)(($t = array_combine($cols, $t2 = str_getcsv($line, $splitchar, '"'))) ? $t : it::error("schema mismatch: " . count($cols) . " vs " . count($t2))); # NOPHPLINT

	setlocale(LC_CTYPE, $oldlocale);

	return (array)$records;
}

/**
 * Return contents of pipe as array of lines
 */
function lines()
{
	return $this->lines;
}

/**
 * Pipe our contents through a shell command
 */
function pipe($cmd)
{
	$descriptors = array(0 => array("pipe", "r"), 1 => array("pipe", "w"));
	$process = proc_open($cmd, $descriptors, $pipes);

	fwrite($pipes[0], join("\n", $this->lines) . ($this->lines ? "\n" :""));
	fclose($pipes[0]);

	$this->lines = explode("\n", rtrim(stream_get_contents($pipes[1])));
	fclose($pipes[1]);
	proc_close($process);

	return $this;
}

/**
 * Save our contents in a file
 * @param $fn filename to save in, omit or null for stdout
 * @param $append append to file (boolean)
 */
function write($fn = null, $append = false)
{
	file_put_contents($fn === null ? "php://stdout" : $fn, $this->lines ? join("\n", $this->lines) . "\n" : "", $append ? FILE_APPEND : 0);

	return $this;
}

}