<commit_before>"""Lazily-evaluated stream with pipelining via the '>>' operator.

Introduction
============

Streams are generalized iterators with a pipelining mechanism to enable
data-flow programming.

The idea is to take the output of a function that turn an iterable into
another iterable and plug that as the input of another such function.
While you can already do this using function composition, this package
provides an elegant notation for it by overloading the '>>' operator.

This approach focuses the programming on processing streams of data, step
by step.  A pipeline usually starts with a producer, then passes through
a number of filters.  Multiple streams can be branched and combined.
Finally, the output is fed to an accumulator, which can be any function
of one iterable argument.

**Producers**:  anything iterable
	+ from this module:  seq, gseq, repeatcall, chaincall

**Filters**:
	+ by index:  take, drop, takei, dropi
	+ by condition:  filter, takewhile, dropwhile
	+ by transformation:  apply, map, fold
	+ by combining streams:  prepend, tee
	+ for special purpose:  chop, cut, flatten

**Accumulators**:  item, maximum, minimum, reduce
	+ from Python:  list, sum, dict, max, min ...

Values are computed only when an accumulator forces some or all evaluation
(not when the stream are set up).


Parallelization
===============

When a producer is doing blocking I/O, it is possible to use a ThreadedFeeder
or ForkedFeeder to improve performance.  The feeder will start a thread or a
process to run the producer and feed generated items back to the pipeline, thus
minimizing the time that the whole pipeline has to wait when the producer is
blocking in system calls.


Articles
========

Articles written about this module by the author can be retrieved from
<http://blog.onideas.ws/project:stream.py>.


Examples
========

Better itertools.islice
-----------------------
::

  >>> from itertools import count
  >>> c = count()
  >>> c >> item[1:10:2]
  [1, 3, 5, 7, 9]
  >>> c >> item[:5]
  [10, 11, 12, 13, 14]

String processing
-----------------
Grep some lines matching a regex from a file, cut out the 4th field
separated by ' ', ':' or '.', then save as a list::

    import re
    s = open('file') \
      >> filter(re.compile(regex).search) \
      >> map(re.compile(' |:|\.').split) \
      >> map(itemgetter(3)) \
      >> list

Partial sums
------------
Compute the first few partial sums of the geometric series 1 + 1/2 + 1/4 + ..::

    >>> gseq(0.5) >> fold(operator.add) >> item[:5]
    [1, 1.5, 1.75, 1.875, 1.9375]

Random Walk in 2D
-----------------
Generate an infinite stream of coordinates representing the position of
a random walker in 2D::

    from random import choice
    vectoradd = lambda u,v: zip(u, v) >> apply(operator.add) >> list
    directions = [[1,0], [0,1], [-1,0], [0,-1]]
    rw = lambda: repeatcall(choice, directions) >> fold(vectoradd, [0, 0])

Calling choice repeatedly yields the series of unit vectors representing the
directions that the walker takes, then these vectors are gradually added to get
a series of coordinates.

To instantiate a random-walk, and get the first 10 coordinates::

    walk = rw()
    walk >> item[:10]

Question: what is the farthest point that the walker wanders upto the first
return to the origin? (Note that he might never return at all!)::

    vectorlen = lambda v: v >> map(lambda x: x**2) >> sum
    rw() >> drop(1) >> takewhile(lambda v: v != [0, 0]) >> maximum(key=vectorlen)

The first coordinate [0, 0], which is the origin, needs to be dropped otherwise
takewhile will truncate immediately.

We can also probe into the walker's chosen path::

    probe = Stream()
    rw() >> drop(1) >> takewhile(lambda v: v != [0, 0]) >> tee(probe) >> maximum(key=vectorlen)

Now you can see his exact coordinates, for example the first 10 are::

    probe >> item[:10]
"""

__version__ = '0.8'


import __builtin__
import copy
import collections
import heapq
import itertools
import multiprocessing
import operator
import platform
import Queue
import re
import select
import threading
import time

from operator import itemgetter, attrgetter

try:
	from operator import methodcaller
except ImportError:
	def methodcaller(method, *args, **kwargs):
		return lambda o: o.method(*args, **kwargs)

zip = itertools.izip


#_____________________________________________________________________
# Base class


class BrokenPipe(Exception):
	pass


class Stream(collections.Iterable):
	"""A class representing both a stream and a filter.

	The outgoing stream is represented by the attribute 'iterator'.

	The filter is represented by the method __call__(inpipe), which
	combines self's iterator with inpipe's, returning a new iterator
	representing a new outgoing stream.
	
	A Stream subclass will usually implement __call__, unless it is an
	accumulator and will not return a Stream, in which case it needs to
	implement __pipe__.  The default piping mechanism of Stream is appending
	to the end of the its input (which had better terminate!).

	>>> [1, 2, 3] >> Stream('foo') >> Stream('bar') >> list
	[1, 2, 3, 'f', 'o', 'o', 'b', 'a', 'r']
	"""
	def __init__(self, iterable=None):
		"""Make a stream object from an interable."""
		self.iterator = iter(iterable if iterable else [])

	def __iter__(self):
		return self.iterator

	def __call__(self, inpipe):
		return itertools.chain(inpipe, self.iterator)

	def __pipe__(self, inpipe):
		self.iterator = self.__call__(iter(inpipe))
		return self

	@staticmethod
	def pipe(inpipe, outpipe):
		"""Connect inpipe and outpipe.  If outpipe is not a Stream instance,
		it should be an function callable on an iterable.
		"""
		if hasattr(outpipe, '__pipe__'):
			return outpipe.__pipe__(inpipe)
		elif hasattr(outpipe, '__call__'):
			if hasattr(outpipe, '__name__') and outpipe.__name__ == 'list':
				## For some reason `list` doesn't believe that inpipe is an iterator
				return outpipe(iter(inpipe))
			else:
				return outpipe(inpipe)
		else:
			raise BrokenPipe('No connection mechanism defined')

	def __rshift__(self, outpipe):
		return Stream.pipe(self, outpipe)

	def __rrshift__(self, inpipe):
		return Stream.pipe(inpipe, self)

	def extend(self, inpipe):
		"""Similar to pipe(), except that extend() is an instance method,
		and inpipe must be a Stream, in which case inpipe.iterator is
		modified in-place.
		"""
		inpipe.iterator = self.__call__(inpipe.iterator)
		return inpipe

	def __len__(self):				### this will force all evaluation
		"""
		>>> Stream(range(20)) >> len
		20
		"""
		return len([i for i in self.iterator])

	def __repr__(self):
		return 'Stream(%s)' % repr(self.iterator)


#_______________________________________________________________________
# Filtering streams by element indices


class take(Stream):
	"""Take the firts n items of the input stream, return a Stream.
	
	>>> seq(1, 2) >> take(10)
	Stream([1, 3, 5, 7, 9, 11, 13, 15, 17, 19])
	"""
	def __init__(self, n):
		"""n: the number of elements to be taken"""
		super(take, self).__init__()
		self.n = n<commit_msg>Eureka!  Having __len__ confused the hell out of list(), it hereby removed!<commit_after>"""Lazily-evaluated stream with pipelining via the '>>' operator.

Introduction
============

Streams are generalized iterators with a pipelining mechanism to enable
data-flow programming.

The idea is to take the output of a function that turn an iterable into
another iterable and plug that as the input of another such function.
While you can already do this using function composition, this package
provides an elegant notation for it by overloading the '>>' operator.

This approach focuses the programming on processing streams of data, step
by step.  A pipeline usually starts with a producer, then passes through
a number of filters.  Multiple streams can be branched and combined.
Finally, the output is fed to an accumulator, which can be any function
of one iterable argument.

**Producers**:  anything iterable
	+ from this module:  seq, gseq, repeatcall, chaincall

**Filters**:
	+ by index:  take, drop, takei, dropi
	+ by condition:  filter, takewhile, dropwhile
	+ by transformation:  apply, map, fold
	+ by combining streams:  prepend, tee
	+ for special purpose:  chop, cut, flatten

**Accumulators**:  item, maximum, minimum, reduce
	+ from Python:  list, sum, dict, max, min ...

Values are computed only when an accumulator forces some or all evaluation
(not when the stream are set up).


Parallelization
===============

When a producer is doing blocking I/O, it is possible to use a ThreadedFeeder
or ForkedFeeder to improve performance.  The feeder will start a thread or a
process to run the producer and feed generated items back to the pipeline, thus
minimizing the time that the whole pipeline has to wait when the producer is
blocking in system calls.


Articles
========

Articles written about this module by the author can be retrieved from
<http://blog.onideas.ws/project:stream.py>.


Examples
========

Better itertools.islice
-----------------------
::

  >>> from itertools import count
  >>> c = count()
  >>> c >> item[1:10:2]
  [1, 3, 5, 7, 9]
  >>> c >> item[:5]
  [10, 11, 12, 13, 14]

String processing
-----------------
Grep some lines matching a regex from a file, cut out the 4th field
separated by ' ', ':' or '.', then save as a list::

    import re
    s = open('file') \
      >> filter(re.compile(regex).search) \
      >> map(re.compile(' |:|\.').split) \
      >> map(itemgetter(3)) \
      >> list

Partial sums
------------
Compute the first few partial sums of the geometric series 1 + 1/2 + 1/4 + ..::

    >>> gseq(0.5) >> fold(operator.add) >> item[:5]
    [1, 1.5, 1.75, 1.875, 1.9375]

Random Walk in 2D
-----------------
Generate an infinite stream of coordinates representing the position of
a random walker in 2D::

    from random import choice
    vectoradd = lambda u,v: zip(u, v) >> apply(operator.add) >> list
    directions = [[1,0], [0,1], [-1,0], [0,-1]]
    rw = lambda: repeatcall(choice, directions) >> fold(vectoradd, [0, 0])

Calling choice repeatedly yields the series of unit vectors representing the
directions that the walker takes, then these vectors are gradually added to get
a series of coordinates.

To instantiate a random-walk, and get the first 10 coordinates::

    walk = rw()
    walk >> item[:10]

Question: what is the farthest point that the walker wanders upto the first
return to the origin? (Note that he might never return at all!)::

    vectorlen = lambda v: v >> map(lambda x: x**2) >> sum
    rw() >> drop(1) >> takewhile(lambda v: v != [0, 0]) >> maximum(key=vectorlen)

The first coordinate [0, 0], which is the origin, needs to be dropped otherwise
takewhile will truncate immediately.

We can also probe into the walker's chosen path::

    probe = Stream()
    rw() >> drop(1) >> takewhile(lambda v: v != [0, 0]) >> tee(probe) >> maximum(key=vectorlen)

Now you can see his exact coordinates, for example the first 10 are::

    probe >> item[:10]
"""

__version__ = '0.8'


import __builtin__
import copy
import collections
import heapq
import itertools
import multiprocessing
import operator
import platform
import Queue
import re
import select
import threading
import time

from operator import itemgetter, attrgetter

try:
	from operator import methodcaller
except ImportError:
	def methodcaller(method, *args, **kwargs):
		return lambda o: o.method(*args, **kwargs)

zip = itertools.izip


#_____________________________________________________________________
# Base class


class BrokenPipe(Exception):
	pass


class Stream(collections.Iterable):
	"""A class representing both a stream and a filter.

	The outgoing stream is represented by the attribute 'iterator'.

	The filter is represented by the method __call__(inpipe), which
	combines self's iterator with inpipe's, returning a new iterator
	representing a new outgoing stream.
	
	A Stream subclass will usually implement __call__, unless it is an
	accumulator and will not return a Stream, in which case it needs to
	implement __pipe__.  The default piping mechanism of Stream is appending
	to the end of the its input (which had better terminate!).

	>>> [1, 2, 3] >> Stream('foo') >> Stream('bar') >> list
	[1, 2, 3, 'f', 'o', 'o', 'b', 'a', 'r']
	"""
	def __init__(self, iterable=None):
		"""Make a stream object from an interable."""
		self.iterator = iter(iterable if iterable else [])

	def __iter__(self):
		return self.iterator

	def __call__(self, inpipe):
		return itertools.chain(inpipe, self.iterator)

	def __pipe__(self, inpipe):
		self.iterator = self.__call__(iter(inpipe))
		return self

	@staticmethod
	def pipe(inpipe, outpipe):
		"""Connect inpipe and outpipe.  If outpipe is not a Stream instance,
		it should be an function callable on an iterable.
		"""
		if hasattr(outpipe, '__pipe__'):
			return outpipe.__pipe__(inpipe)
		elif hasattr(outpipe, '__call__'):
			return outpipe(inpipe)
		else:
			raise BrokenPipe('No connection mechanism defined')

	def __rshift__(self, outpipe):
		return Stream.pipe(self, outpipe)

	def __rrshift__(self, inpipe):
		return Stream.pipe(inpipe, self)

	def extend(self, inpipe):
		"""Similar to pipe(), except that extend() is an instance method,
		and inpipe must be a Stream, in which case inpipe.iterator is
		modified in-place.
		"""
		inpipe.iterator = self.__call__(inpipe.iterator)
		return inpipe

	def __repr__(self):
		return 'Stream(%s)' % repr(self.iterator)


#_______________________________________________________________________
# Filtering streams by element indices


class take(Stream):
	"""Take the firts n items of the input stream, return a Stream.
	
	>>> seq(1, 2) >> take(10)
	Stream([1, 3, 5, 7, 9, 11, 13, 15, 17, 19])
	"""
	def __init__(self, n):
		"""n: the number of elements to be taken"""
		super(take, self).__init__()
		self.n = n
{'fraction_non_alphanumeric': 0.11615979195261143, 'fraction_numerical': 0.014881167377013654, 'mean_word_length': 3.940756602426838, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 164, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1001418', 'n_tokens_mistral': 4343, 'n_tokens_neox': 4027, 'n_words': 1734}