iter Module Reference

flatten(xs)

Single-level list flattening

Source code in haskellian/src/haskellian/iter/basics.py

@I.lift
def flatten(xs: Iterable[Iterable[A]]) -> Iterable[A]:
  """Single-level list flattening"""
  return chain.from_iterable(xs)

isiterable(x, str_iterable=False, bytes_iterable=False)

Is x iterable? - str_iterable, bytes_iterable: whether str and bytes are considered iterable

Source code in haskellian/src/haskellian/iter/basics.py

def isiterable(x, str_iterable = False, bytes_iterable = False) -> bool:
  """Is `x` iterable?
  - `str_iterable`, `bytes_iterable`: whether `str` and `bytes` are considered iterable
  """
  return isinstance(x, Iterable) and \
    (not isinstance(x, str) or str_iterable) and \
    (not isinstance(x, bytes) or bytes_iterable)

range(start=0, end=None, step=1)

Like range, but possibly infinite (if end is None)

Source code in haskellian/src/haskellian/iter/basics.py

@I.lift
def range(start: int = 0, end: int | None = None, step: int = 1) -> Iterable[int]:
  """Like `range`, but possibly infinite (if `end is None`)"""
  i = start
  while end is None or i < end:
    yield i
    i += step

batch(n, xs)

Batches xs into n-tuples

Source code in haskellian/src/haskellian/iter/batching.py

@I.lift
def batch(n: int, xs: Iterable[A]) -> Iterable[tuple[A, ...]]:
  """Batches `xs` into `n`-tuples"""
  it = iter(xs)
  while b := tuple(itertools.islice(it, n)):
    yield b

lazy_batch(n, xs)

Batches xs into n-tuples, but returns lazy iterators

Source code in haskellian/src/haskellian/iter/batching.py

@I.lift
def lazy_batch(n: int, xs: Iterable[A]) -> Iterable[Iter[A]]:
  """Batches `xs` into `n`-tuples, but returns lazy iterators"""
  it = iter(xs)
  while not (b := I.take(n, it)).empty():
    yield b

lazy_shard(min_size, size, xs)

Shards xs into groups of at least min_size based on size (last shard may have less)

Source code in haskellian/src/haskellian/iter/batching.py

@I.lift
def lazy_shard(min_size: float, size: Callable[[A], float], xs: Iterable[A]) -> Iterable[Iter[A]]:
  """Shards `xs` into groups of at least `min_size` based on `size` (last shard may have less)"""
  it = iter(xs)
  @I.lift
  def shard():
    acc = 0
    for x in it:
      acc += size(x)
      yield x
      if acc >= min_size:
        break
  while not (s := shard()).empty():
    yield s

shard(min_size, size, xs)

Shards xs into groups of at least min_size based on size (last shard may have less)

Source code in haskellian/src/haskellian/iter/batching.py

@I.lift
def shard(min_size: float, size: Callable[[A], float], xs: Iterable[A]) -> Iterable[list[A]]:
  """Shards `xs` into groups of at least `min_size` based on `size` (last shard may have less)"""
  shard = []
  acc = 0
  for x in xs:
    acc += size(x)
    shard.append(x)
    if acc >= min_size:
      yield shard
      shard = []
      acc = 0
  if shard:
    yield shard

split(n, xs)

Splits xs into a list of the first n and a generator of the rest - e.g: split(3, [1,2,3,4,5]) == ([1, 2, 3], generator(4, 5))

Source code in haskellian/src/haskellian/iter/batching.py

def split(n: int, xs: Iterable[A]) -> tuple[list[A], Iterable[A]]:
  """Splits `xs` into a list of the first `n` and a generator of the rest
  - e.g: `split(3, [1,2,3,4,5]) == ([1, 2, 3], generator(4, 5))`"""
  it = iter(xs)
  init = I.take(n, it).sync()
  return init, (x for x in it)

at(i, xs)

Safe list indexing

Source code in haskellian/src/haskellian/iter/indexing.py

def at(i: int, xs: list[int]) -> int | None:
  """Safe list indexing"""
  try:
    return xs[i]
  except IndexError:
    ...

pick(indices, xs)

Pick xs values at indices indices - E.g. pick([1, 2, 4], ['a', 'b', 'c', 'd', 'e', 'f', 'g']) = ['b', 'c', 'e']

Source code in haskellian/src/haskellian/iter/indexing.py

def pick(indices: Iterable[int], xs: list[A]) -> list[A]:
  """Pick `xs` values at indices `indices`
  - E.g. `pick([1, 2, 4], ['a', 'b', 'c', 'd', 'e', 'f', 'g']) = ['b', 'c', 'e']`"""
  return [xs[i] for i in indices]

lift(func)

Lift an iterable function func to return an Iter

Source code in haskellian/src/haskellian/iter/lifting.py

def lift(func: Callable[P, Iterable[A]]) -> Callable[P, I.Iter[A]]:
  """Lift an iterable function `func` to return an `Iter`"""
  @wraps(func)
  def _f(*args: P.args, **kwargs: P.kwargs) -> I.Iter[A]:
    return I.Iter(func(*args, **kwargs))
  return _f

pluck(xs, key)

Extract values by key. Skips dicts without it.

Source code in haskellian/src/haskellian/iter/maps.py

def pluck(xs: Iterable[Mapping[A, B]], key: A) -> Iterable[B]:
  """Extract values by key. Skips dicts without it."""
  for x in xs:
    if key in x:
      yield x[key]

oversample(data)

Balance lazy iterators by repeating the shorter ones

Source code in haskellian/src/haskellian/iter/misc.py

@I.lift
def oversample(data: Sequence[Callable[[], Iterable[A]]]) -> Iterable[A]:
  """Balance lazy iterators by repeating the shorter ones"""
  iters = [iter(repeat(d)) for d in data]
  while True:
    for i in iters:
      yield next(i)

repeat(iter)

Repeat a lazy iterable indefinitely

Source code in haskellian/src/haskellian/iter/misc.py

@I.lift
def repeat(iter: Callable[[], Iterable[A]]) -> Iterable[A]:
  """Repeat a lazy iterable indefinitely"""
  while True:
    yield from iter()

shuffle(xs, shuffle_size)

Reservoir sampling based shuffling

Source code in haskellian/src/haskellian/iter/misc.py

@I.lift
def shuffle(xs: Iterable[A], shuffle_size: int) -> Iterable[A]:
  """Reservoir sampling based shuffling"""
  reservoir = []
  for x in xs:
    if len(reservoir) < shuffle_size:
      reservoir.append(x)
    else:
      i = random.randrange(0, shuffle_size)
      yield reservoir[i]
      reservoir[i] = x

  random.shuffle(reservoir)
  yield from reservoir

undersample(data)

Balance lazy iterators by truncating the longer ones

Source code in haskellian/src/haskellian/iter/misc.py

@I.lift
def undersample(data: Sequence[Callable[[], Iterable[A]]]) -> Iterable[A]:
  """Balance lazy iterators by truncating the longer ones"""
  while True:
    for t in zip(*[d() for d in data]):
      yield from t

ndenumerate(xxs, depth=None)

ndenumerate(xxs: Iterable[Iterable[A]], depth: Literal[2] | None = None) -> I.Iter[tuple[tuple[int, int], A]]

ndenumerate(xxs: Iterable[Iterable[Iterable[A]]], depth: Literal[3] | None = None) -> I.Iter[tuple[tuple[int, int, int], A]]

ndenumerate(xxs: Iterable, depth: int | None = None) -> I.Iter[tuple[tuple[int, ...], Any]]

Like enumerate, but for nested iterables.

So, instead of

for i, xxs in enumerate(xxxs):
for j, xs in enumerate(xxs):
  for k, x in enumerate(xs):
  ...

You can just do

for x, (i, j, k) in ndenumerate(xxxs):
...

Note: - By default it unrolls all the way down until a none-iterable object is found. - If you want to iterate over iterables (e.g. lists, numpy arrays, tensors), you'll have to pass in a fixed depth

Source code in haskellian/src/haskellian/iter/nested.py

@I.lift
def ndenumerate(xxs, depth: int | None = None): # type: ignore
  """Like `enumerate`, but for nested iterables.

  So, instead of

  ```
  for i, xxs in enumerate(xxxs):
  for j, xs in enumerate(xxs):
    for k, x in enumerate(xs):
    ...
  ```

  You can just do

  ```
  for x, (i, j, k) in ndenumerate(xxxs):
  ...
  ```

  Note:
  - By default it unrolls all the way down until a none-iterable object is found.
  - If you want to iterate over iterables (e.g. lists, numpy arrays, tensors), you'll have to pass in a fixed `depth`
  """
  return _auto_ndenumerate(xxs) if depth is None else _fixed_ndenumerate(xxs, depth)

ndflat(xxs, depth=None)

ndflat(xxs: Iterable[Iterable[A]], depth: Literal[2] | None = None) -> I.Iter[I.Iter[A]]

ndflat(xxs: Iterable[Iterable[Iterable[A]]], depth: Literal[3] | None = None) -> I.Iter[A]

ndflat(xxs: Iterable, depth: int | None = None) -> I.Iter

ndflat([[x1, x2], [x3, [x4]]]) = [x1, x2, x3, x4]

Source code in haskellian/src/haskellian/iter/nested.py

@I.lift
def ndflat(xxs: Iterable[Iterable[A]], depth: int | None = None): # type: ignore
  """`ndflat([[x1, x2], [x3, [x4]]]) = [x1, x2, x3, x4]`"""
  return _auto_ndflat(xxs) if depth is None else _fixed_ndflat(xxs, depth) # type: ignore

ndrange(*ranges)

ndrange(rng1: Range) -> I.Iter[tuple[int]]

ndrange(rng1: Range, rng2: Range) -> I.Iter[tuple[int, int]]

ndrange(rng1: Range, rng2: Range, rng3: Range) -> I.Iter[tuple[int, int, int]]

ndrange(*ranges: Range) -> I.Iter[tuple[int, ...]]

Like range, but returns an iterable of n-tuples instead of ints.

So, instead of

for i in range(m):
for j in range(n):
  ...

You can just do:

for i, j in ndrange(m, n):
...

And it also supports normal (start, end) or (start, end, step) ranges:

for i, j in ndrange((-10, 0, 2), (4, 6)):
...
# (-10, 4), (-10, 5), (-8, 4), (-8, 5), ...

Source code in haskellian/src/haskellian/iter/nested.py

@I.lift
def ndrange(*ranges): # type: ignore
  """Like `range`, but returns an iterable of `n`-tuples instead of ints.

  So, instead of
  ```
  for i in range(m):
  for j in range(n):
    ...
  ```
  You can just do:
  ```
  for i, j in ndrange(m, n):
  ...
  ```

  And it also supports normal `(start, end)` or `(start, end, step)` ranges:
  ```
  for i, j in ndrange((-10, 0, 2), (4, 6)):
  ...
  # (-10, 4), (-10, 5), (-8, 4), (-8, 5), ...
  ```
  """
  def _ndrange(*ranges, inds = ()):
    if len(ranges) == 0:
      yield inds
    else:
      [r, *rs] = ranges
      rng = range(*r) if isiterable(r) else range(r)
      for i in rng:
        yield from _ndrange(*rs, inds=inds+(i,))

  yield from _ndrange(*ranges)

every(n, xs)

Take every nth element of xs - every(3, range(10)) == Iter([0, 3, 6, 9])

Source code in haskellian/src/haskellian/iter/slicing.py

@I.lift
def every(n: int, xs: Iterable[A]) -> Iterable[A]:
	"""Take every `n`th element of `xs`
	- `every(3, range(10)) == Iter([0, 3, 6, 9])`
	"""
	for i, x in enumerate(xs):
		if i % n == 0:
			yield x

fst(t)

fst((a, _)) = a

Source code in haskellian/src/haskellian/iter/slicing.py

def fst(t: tuple[A, Unpack[As]]) -> A:
	"""`fst((a, _)) = a`"""
	x, *_ = t
	return x

head(xs)

head([x, *_]) = x

Source code in haskellian/src/haskellian/iter/slicing.py

def head(xs: Iterable[A]) -> A | None:
  """`head([x, *_]) = x`"""
  for x in xs:
    return x

last(xs)

head([x, *_]) = x

Source code in haskellian/src/haskellian/iter/slicing.py

def last(xs: Iterable[A]) -> A | None:
  """`head([x, *_]) = x`"""
  ys = list(xs)
  if len(ys) > 0:
    return ys[-1]

pad(n, fill, xs)

Pads xs to length n, appending fills as needed

Source code in haskellian/src/haskellian/iter/slicing.py

@I.lift
def pad(n: int, fill: B, xs: Iterable[A]) -> Iterable[A|B]:
	"""Pads `xs` to length `n`, appending `fill`s as needed"""
	i = 0
	for i, x in enumerate(xs):
		yield x

	for _ in range(i+1, n):
		yield fill

skip(n, xs)

skip(n, [x1, ..., xn, *xs]) = xs

Source code in haskellian/src/haskellian/iter/slicing.py

@I.lift  
def skip(n: int, xs: Iterable[A]) -> Iterable[A]:
	"""`skip(n, [x1, ..., xn, *xs]) = xs`"""
	return itertools.islice(xs, n, None)

snd(t)

snd((_, b)) = b

Source code in haskellian/src/haskellian/iter/slicing.py

def snd(t: tuple[A, B, Unpack[As]]) -> B:
	"""`snd((_, b)) = b`"""
	_, x, *_ = t
	return x

tail(xs)

tail([_, *xs]) = xs

Source code in haskellian/src/haskellian/iter/slicing.py

@I.lift
def tail(xs: Iterable[A]) -> Iterable[A]:
	"""`tail([_, *xs]) = xs`"""
	return skip(1, xs)

take(n, xs)

take(n, [x1, ..., xn, *_]) = [x1, ..., xn]

Source code in haskellian/src/haskellian/iter/slicing.py

@I.lift
def take(n: int | None, xs: Iterable[A]) -> Iterable[A]:
	"""`take(n, [x1, ..., xn, *_]) = [x1, ..., xn]`"""
	return itertools.islice(xs, n)

transpose(xs)

Transpose a 2d list, cropping to the shortest row. E.g:

transpose([
    [1, 2],
    [3, 4],
    [5, 6]
]) == [
    [1, 3, 5],
    [2, 4, 6]
]

but

transpose([
    [1, 2],
    [3, 4],
    [5]
]) == [
    [1, 2, 3],
]

Source code in haskellian/src/haskellian/iter/transposing.py

def transpose(xs: Iterable[Iterable[A]]) -> list[list[A]]:
    """Transpose a 2d list, cropping to the shortest row. E.g:
    ```
    transpose([
        [1, 2],
        [3, 4],
        [5, 6]
    ]) == [
        [1, 3, 5],
        [2, 4, 6]
    ]
    ```
    but
    ```
    transpose([
        [1, 2],
        [3, 4],
        [5]
    ]) == [
        [1, 2, 3],
    ]
    ```
    """
    return list(zip(*xs))

transpose_ragged(xs)

Like transpose, but resulting rows can be ragged (i.e. have different lengths). E.g:

transpose([
    [1, 2],
    [3, 4],
    [5]
]) == [
    [1, 3, 5],
    [2, 4]
]

Source code in haskellian/src/haskellian/iter/transposing.py

def transpose_ragged(xs: Iterable[Iterable[A]]) -> list[list[A]]:
    """Like `transpose`, but resulting rows can be ragged (i.e. have different lengths). E.g:
    ```
    transpose([
        [1, 2],
        [3, 4],
        [5]
    ]) == [
        [1, 3, 5],
        [2, 4]
    ]
    ```"""
    return [[x for x in cols if x is not None] for cols in zip_longest(*xs)]

interleave(xs)

Interleave multiple iterators based on their weight. - (weight, iter) = xs[i]: iter is yielded weight times before moving to the next iterator.

Source code in haskellian/src/haskellian/iter/zipping.py

@I.lift
def interleave(xs: Sequence[tuple[int, Iterable[A]]]) -> Iterable[A]:
  """Interleave multiple iterators based on their weight.
  - `(weight, iter) = xs[i]`: `iter` is yielded `weight` times before moving to the next iterator.
  """
  weights, iters = I.unzip(xs)
  iters = [iter(it) for it in iters]
  weights = list(weights)
  while True:
    for i, (w, it) in enumerate(zip(weights, iters)):
      try:
        for _ in range(w):
          yield next(it)
      except StopIteration:
        weights.pop(i)
        iters.pop(i)
        if not weights:
          return

pairwise(xs)

pairwise([x1, x2, x3, x4, ...]) = [(x1, x2), (x2, x3), (x3, x4), ...] - pairwise([]) = [] - pairwise([x]) = []

Source code in haskellian/src/haskellian/iter/zipping.py

@I.lift
def pairwise(xs: Iterable[A]) -> Iterable[tuple[A, A]]:
    """`pairwise([x1, x2, x3, x4, ...]) = [(x1, x2), (x2, x3), (x3, x4), ...]`
    - `pairwise([]) = []`
    - `pairwise([x]) = []`
    """
    x0, tail = uncons(xs)
    for x1 in tail:
        yield x0, x1 # type: ignore (`x0` won't be `None` if there's some element in `tail`)
        x0 = x1

uncons(xs)

uncons([x, *xs]) = (x, xs)

Source code in haskellian/src/haskellian/iter/zipping.py

def uncons(xs: Iterable[A]) -> tuple[A | None, I.Iter[A]]:
    """`uncons([x, *xs]) = (x, xs)`"""
    it = iter(xs)
    x = next(it, None)
    return x, I.Iter(it)

unzip(xs)

unzip(xs: Iterable[tuple[A, B]]) -> tuple[list[A], list[B]]

unzip(xs: Iterable[tuple[A, B, C]]) -> tuple[list[A], list[B], list[C]]

unzip(xs: Iterable[tuple[A, B, C, D]]) -> tuple[list[A], list[B], list[C], list[D]]

unzip(xs: Iterable[tuple[A, B, C, D, E]]) -> tuple[list[A], list[B], list[C], list[D], list[E]]

unzip(xs: Iterable[tuple[A, B, C, D, E, F]]) -> tuple[list[A], list[B], list[C], list[D], list[E], list[F]]

unzip(xs: Iterable[tuple[A, ...]]) -> tuple[list[A], ...]

[(a, b, ...)] -> ([a], [b], ...)

Source code in haskellian/src/haskellian/iter/zipping.py

def unzip(xs):
    """`[(a, b, ...)] -> ([a], [b], ...)`"""
    return tuple(map(list, zip(*xs))) # type: ignore