Skip to content

Iterator Tools Implementation

monad_std.iter.impl.default_iter._IterIterator

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/default_iter.py 
20
21
22
23
24
25
26
27
class _IterIterator(IterMeta[T], t.Generic[T]):
    __iter: t.Iterator[T]

    def __init__(self, v: t.Iterator[T]):
        self.__iter = v

    def next(self) -> Option[T]:
        return Result.catch(self.__iter.__next__).ok()

monad_std.iter.impl.default_iter._IterIterable

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/default_iter.py 
10
11
12
13
14
15
16
17
class _IterIterable(IterMeta[T], t.Generic[T]):
    __iter: t.Iterator[T]

    def __init__(self, v: t.Iterable[T]):
        self.__iter = iter(v)

    def next(self) -> Option[T]:
        return Result.catch(self.__iter.__next__).ok()

monad_std.iter.impl.default_iter._Iter

Bases: Iterator[T], Generic[T]

Source code in monad_std/iter/impl/default_iter.py 
30
31
32
33
34
35
36
37
38
39
40
41
class _Iter(t.Iterator[T], t.Generic[T]):
    __iter: IterMeta[T]

    def __init__(self, v: IterMeta[T]):
        self.__iter = v

    def __next__(self):
        n = self.__iter.next()
        if n.is_none():
            raise StopIteration
        else:
            return n.unwrap_unchecked()

monad_std.iter.impl.array_chunk.ArrayChunk

Bases: IterMeta[List[T]], Generic[T]

Source code in monad_std/iter/impl/array_chunk.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class ArrayChunk(IterMeta[t.List[T]], t.Generic[T]):
    __it: IterMeta[T]
    __chunk_size: int
    __unused: Option[t.List[T]]

    def __init__(self, it: IterMeta[T], chunk_size: int):
        assert chunk_size > 0, "Chunk size must be greater than zero!"
        self.__it = it
        self.__chunk_size = chunk_size

    def next(self) -> Option[t.List[T]]:
        arr = []
        for _ in range(self.__chunk_size):
            if (x := self.__it.next()).is_some():
                arr.append(x.unwrap())
            else:
                break
        else:
            # short-circuit here if we don't meet any Option::None
            return Option.some(arr)
        self.__unused = Option.some(arr)
        return Option.none()

    def get_unused(self) -> Option[t.List[T]]:
        """Return the last/unused several elements.

        Examples:
            ```python
            it = IterMeta.iter([1, 2, 3, 4]).array_chunk(3)
            assert it.next() == Option.some([1, 2, 3])
            assert it.next() == Option.none()
            assert it.get_unused() == Option.some([4])
            ```
        """
        return self.__unused

get_unused()

Return the last/unused several elements.

Examples:

1
2
3
4
it = IterMeta.iter([1, 2, 3, 4]).array_chunk(3)
assert it.next() == Option.some([1, 2, 3])
assert it.next() == Option.none()
assert it.get_unused() == Option.some([4])
Source code in monad_std/iter/impl/array_chunk.py 
33
34
35
36
37
38
39
40
41
42
43
44
def get_unused(self) -> Option[t.List[T]]:
    """Return the last/unused several elements.

    Examples:
        ```python
        it = IterMeta.iter([1, 2, 3, 4]).array_chunk(3)
        assert it.next() == Option.some([1, 2, 3])
        assert it.next() == Option.none()
        assert it.get_unused() == Option.some([4])
        ```
    """
    return self.__unused

monad_std.iter.impl.batch.Batching

Bases: IterMeta[B], Generic[It, B]

Source code in monad_std/iter/impl/batch.py 
10
11
12
13
14
15
16
17
18
19
class Batching(IterMeta[B], t.Generic[It, B]):
    __it: It
    __func: t.Callable[[It], Option[B]]

    def __init__(self, __it: It, __func: t.Callable[[It], Option[B]]):
        self.__it = __it
        self.__func = __func

    def next(self) -> Option[B]:
        return self.__func(self.__it)

monad_std.iter.impl.chain.Chain

Bases: IterMeta[T], Generic[T, It1, It2]

Source code in monad_std/iter/impl/chain.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Chain(IterMeta[T], t.Generic[T, It1, It2]):
    __it1: Option[It1]
    __it2: Option[It2]

    def __init__(self, one: Option[It1], another: Option[It2]):
        self.__it1 = one
        self.__it2 = another

    def __clear_it2(self) -> Option[T]:
        self.__it2 = Option.none()
        return Option.none()

    def __clear_it1(self) -> Option[T]:
        self.__it1 = Option.none()
        return self.__it2.and_then(lambda x: x.next()).or_else(lambda: self.__clear_it2())

    def next(self) -> Option[T]:
        return self.__it1.and_then(lambda x: x.next()).or_else(lambda: self.__clear_it1())

monad_std.iter.impl.chunk.Chunk

Bases: IterMeta[List[T]], Generic[T]

Source code in monad_std/iter/impl/chunk.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Chunk(IterMeta[t.List[T]], t.Generic[T]):
    __it: ArrayChunk[T]
    __finished: bool

    def __init__(self, it: IterMeta[T], chunk_size: int):
        assert chunk_size > 0, "Chunk size must be greater than zero!"
        self.__it = it.array_chunk(chunk_size)
        self.__finished = False

    def next(self) -> Option[t.List[T]]:
        if not self.__finished:
            nxt = self.__it.next()
            if nxt.is_none():
                unused = self.__it.get_unused()
                self.__finished = True
                return unused
            return nxt
        else:
            return Option.none()

monad_std.iter.impl.enumerate.Enumerate

Bases: IterMeta[Tuple[int, T]], Generic[T]

Source code in monad_std/iter/impl/enumerate.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
class Enumerate(IterMeta[t.Tuple[int, T]], t.Generic[T]):
    __it: IterMeta[T]
    __num: int

    def __init__(self, __it: IterMeta[T]):
        self.__it = __it
        self.__num = 0

    def __self_add(self):
        self.__num += 1

    def next(self) -> Option[t.Tuple[int, T]]:
        return self.__it.next().map(lambda x: (self.__num, x)).inspect(lambda _: self.__self_add())

monad_std.iter.impl.filter.Filter

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/filter.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class Filter(IterMeta[T], t.Generic[T]):
    __it: IterMeta[T]
    __func: t.Callable[[T], bool]

    def __init__(self, __it: IterMeta[T], __func: t.Callable[[T], bool]):
        self.__it = __it
        self.__func = __func

    def next(self) -> Option[T]:
        while (x := self.__it.next()).is_some():
            if self.__func(x.unwrap()):
                return x
        else:
            return Option.none()

monad_std.iter.impl.filter_map.FilterMap

Bases: IterMeta[U], Generic[T, U]

Source code in monad_std/iter/impl/filter_map.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class FilterMap(IterMeta[U], t.Generic[T, U]):
    __it: IterMeta[T]
    __func: t.Callable[[T], Option[U]]

    def __init__(self, it: IterMeta[T], func: t.Callable[[T], Option[U]]):
        self.__it = it
        self.__func = func

    def next(self) -> Option[U]:
        while (x := self.__it.next()).is_some():
            if (z := self.__func(x.unwrap_unchecked())).is_some():
                return z
        else:
            return Option.none()

monad_std.iter.impl.flat_map.FlatMap

Bases: IterMeta[U], Generic[T, U]

Source code in monad_std/iter/impl/flat_map.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class FlatMap(IterMeta[U], t.Generic[T, U]):
    __it: IterMeta[T]
    __current_it: Option[IterMeta[U]]
    __func: t.Callable[[T], t.Union[U, IterMeta[U], t.Iterable[U], t.Iterator[U]]]

    def __init__(self, __it: IterMeta[T], __func: t.Callable[[T], t.Union[U, IterMeta[U], t.Iterable[U], t.Iterator[U]]]):
        self.__it = __it
        self.__func = __func
        self.__current_it = Option.none()

    def __it_next(self) -> Option[U]:
        if (_nxt := self.__it.next()).is_some():
            nxt = self.__func(_nxt.unwrap_unchecked())
            # noinspection DuplicatedCode
            if isinstance(nxt, IterMeta):
                self.__current_it = Option.some(nxt)
                return self.next()
            elif isinstance(nxt, (collections.abc.Iterator, collections.abc.Iterable)):
                self.__current_it = Option.some(IterMeta.iter(nxt))
                return self.next()
            else:
                self.__current_it = Option.none()
                return Option.some(nxt)
        else:
            return Option.none()

    def next(self) -> Option[U]:
        if self.__current_it.is_some():
            x = self.__current_it.and_then(lambda s: s.next())
            if x.is_some():
                return x
            else:
                self.__current_it = Option.none()
        return self.__it_next()

monad_std.iter.impl.flatten.Flatten

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/flatten.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class Flatten(IterMeta[T], t.Generic[T]):
    __it: IterMeta[t.Union[T, IterMeta[T], t.Iterable[T], t.Iterator[T]]]
    __current_it: Option[IterMeta[T]]

    def __init__(self, it: IterMeta[t.Union[T, IterMeta[T], t.Iterable[T], t.Iterator[T]]]):
        self.__it = it
        self.__current_it = Option.none()

    def __it_next(self) -> Option[T]:
        if (_nxt := self.__it.next()).is_some():
            nxt = _nxt.unwrap_unchecked()
            if isinstance(nxt, IterMeta):
                self.__current_it = Option.some(nxt)
                return self.next()
            elif isinstance(nxt, (collections.abc.Iterator, collections.abc.Iterable)):
                self.__current_it = Option.some(IterMeta.iter(nxt))
                return self.next()
            else:
                self.__current_it = Option.none()
                return Option.some(nxt)
        else:
            return Option.none()

    def next(self) -> Option[T]:
        if self.__current_it.is_some():
            x = self.__current_it.and_then(lambda s: s.next())
            if x.is_some():
                return x
            else:
                self.__current_it = Option.none()
        return self.__it_next()

monad_std.iter.impl.fuse.Fuse

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/fuse.py 
 9
10
11
12
13
14
15
16
17
18
19
20
class Fuse(IterMeta[T], t.Generic[T]):
    __it: Option[IterMeta[T]]

    def __init__(self, it: IterMeta[T]):
        self.__it = Option.some(it)

    def __finish(self) -> Option[T]:
        self.__it = Option.none()
        return Option.none()

    def next(self) -> Option[T]:
        return self.__it.and_then(lambda x: x.next()).or_else(lambda: self.__finish())

monad_std.iter.impl.group.GroupBy

Bases: IterMeta[Tuple[K, Group[T, K]]], Generic[T, K]

Source code in monad_std/iter/impl/group.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class GroupBy(IterMeta[t.Tuple[K, Group[T, K]]], t.Generic[T, K]):
    __it: Peekable[T, Fuse[T]]
    __current_yielding: Option[Group[T, K]]
    __current_yielding_key: Option[K]
    __predicate: t.Callable[[T], K]

    def __init__(self, __it: IterMeta[T], __predicate: t.Callable[[T], K]):
        self.__it = __it.fuse().peekable()
        self.__current_yielding = Option.none()
        self.__current_yielding_key = Option.none()
        self.__predicate = __predicate

    def _sub_next(self) -> Option[T]:
        if self.__it.peek().is_none():
            return Option.none()
        else:
            nxt = self.__it.peek().unwrap_unchecked()
            key = self.__predicate(nxt)
            if key == self.__current_yielding_key.unwrap_unchecked():
                return self.__it.next()
            else:
                return Option.none()

    def next(self) -> Option[t.Tuple[K, Group[T, K]]]:
        if self.__it.peek().is_none():
            del self.__current_yielding, self.__current_yielding_key
            return Option.none()
        elif self.__current_yielding.is_none() or self.__current_yielding_key.is_none():
            pass
        else:
            if sys.getrefcount(self.__current_yielding.unwrap_unchecked()) == 2:
                while (peek_nxt := self.__it.peek()).is_some():
                    nxt = peek_nxt.unwrap_unchecked()
                    key = self.__predicate(nxt)
                    if key != self.__current_yielding_key.unwrap_unchecked():
                        break
                    else:
                        self.__it.next()
            else:
                cls: t.List[T] = []

                while (peek_nxt := self.__it.peek()).is_some():
                    nxt = peek_nxt.unwrap_unchecked()
                    key = self.__predicate(nxt)
                    if key == self.__current_yielding_key.unwrap_unchecked():
                        cls.append(self.__it.next().unwrap_unchecked())
                    else:
                        break

                # noinspection PyProtectedMember
                self.__current_yielding.unwrap_unchecked()._extend(cls)

            self.__current_yielding = Option.none()
            self.__current_yielding_key = Option.none()

        peek_nxt = self.__it.peek()
        if peek_nxt.is_none():
            del self.__it  # the inner iterator is fused and can be safely deleted.
            return Option.none()
        key = self.__predicate(peek_nxt.unwrap_unchecked())
        self.__current_yielding_key = Option.some(key)
        group = Group(self)
        self.__current_yielding = Option.some(group)
        return Option.some((key, group))

monad_std.iter.impl.group.Group

Bases: IterMeta[T], Generic[T, K]

Source code in monad_std/iter/impl/group.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class Group(IterMeta[T], t.Generic[T, K]):
    __buffer: Option[IterMeta[T]]
    __parent: "GroupBy[T, K]"
    __end: bool

    def __init__(self, __parent: "GroupBy[T, K]"):
        self.__parent = __parent
        self.__buffer = Option.none()
        self.__end = False

    def _extend(self, el: t.List[T]):
        self.__buffer = Option.some(IterMeta.iter(el))

    def next(self) -> Option[T]:
        if self.__end:
            return Option.none()

        if self.__buffer.is_some():
            return self.__buffer.unwrap_unchecked().next()
        else:
            # noinspection PyProtectedMember
            nxt = self.__parent._sub_next()
            if nxt.is_some():
                return nxt
            else:
                self.__end = True
                return Option.none()

monad_std.iter.impl.inspect.Inspect

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/inspect.py 
 9
10
11
12
13
14
15
16
17
18
class Inspect(IterMeta[T], t.Generic[T]):
    __it: IterMeta[T]
    __func: t.Callable[[T], None]

    def __init__(self, it: IterMeta[T], func: t.Callable[[T], None]):
        self.__it = it
        self.__func = func

    def next(self) -> Option[T]:
        return self.__it.next().inspect(self.__func)

monad_std.iter.impl.intersperse.Intersperse

Bases: IterMeta[T], Generic[T, It]

Source code in monad_std/iter/impl/intersperse.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class Intersperse(IterMeta[T], t.Generic[T, It]):
    __it: Peekable[T, It]
    __sep: T
    __need_sep: bool

    def __init__(self, it: It, sep: T):
        self.__it = it.peekable()
        self.__sep = sep
        self.__need_sep = False

    def next(self) -> Option[T]:
        if self.__need_sep and self.__it.peek().is_some():
            self.__need_sep = False
            return Option.some(self.__sep)
        else:
            self.__need_sep = True
            return self.__it.next()

monad_std.iter.impl.intersperse.IntersperseWith

Bases: IterMeta[T], Generic[T, It]

Source code in monad_std/iter/impl/intersperse.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class IntersperseWith(IterMeta[T], t.Generic[T, It]):
    __it: Peekable[T, It]
    __sep: t.Callable[[], T]
    __need_sep: bool

    def __init__(self, it: It, sep: t.Callable[[], T]):
        self.__it = it.peekable()
        self.__sep = sep
        self.__need_sep = False

    def next(self) -> Option[T]:
        if self.__need_sep and self.__it.peek().is_some():
            self.__need_sep = False
            return Option.some(self.__sep())
        else:
            self.__need_sep = True
            return self.__it.next()

monad_std.iter.impl.map.Map

Bases: IterMeta[U], Generic[T, U]

Source code in monad_std/iter/impl/map.py 
14
15
16
17
18
19
20
21
22
23
class Map(IterMeta[U], t.Generic[T, U]):
    __it: IterMeta[T]
    __func: t.Callable[[T], U]

    def __init__(self, __it: IterMeta[T], __func: t.Callable[[T], U]):
        self.__it = __it
        self.__func = __func

    def next(self) -> Option[U]:
        return self.__it.next().map(self.__func)

monad_std.iter.impl.map.MapWhile

Bases: IterMeta[U], Generic[T, U]

Source code in monad_std/iter/impl/map.py 
26
27
28
29
30
31
32
33
34
35
class MapWhile(IterMeta[U], t.Generic[T, U]):
    __it: IterMeta[T]
    __func: t.Callable[[T], Option[U]]

    def __init__(self, __it: IterMeta[T], __func: t.Callable[[T], Option[U]]):
        self.__it = __it
        self.__func = __func

    def next(self) -> Option[U]:
        return self.__it.next().and_then(lambda v: self.__func(v))

monad_std.iter.impl.map.MapWindows

Bases: IterMeta[R], Generic[T, R]

Source code in monad_std/iter/impl/map.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class MapWindows(IterMeta[R], t.Generic[T, R]):
    __const_len: int
    __it: Option[IterMeta[T]]
    __buffer: Option[t.Deque[T]]
    __func: t.Callable[[t.Deque[T]], R]

    def __init__(self, __const_len: int, __it: IterMeta[T], __func: t.Callable[[t.Deque[T]], R]):
        assert __const_len >= 1, "window size must be larger than 1"
        self.__const_len = __const_len
        self.__it = Option.some(__it)
        self.__func = __func
        self.__buffer = Option.some(collections.deque(maxlen=__const_len))

    def __push_window(self):
        if self.__buffer.is_some() and self.__it.is_some():
            buf = self.__buffer.unwrap_unchecked()
            if len(buf) < self.__const_len:
                while len(buf) < self.__const_len\
                        and (x := self.__it.unwrap_unchecked().next()).is_some():
                    buf.append(x.unwrap_unchecked())
                if len(buf) < self.__const_len:
                    self.__buffer = Option.none()
                    self.__it = Option.none()
            else:
                if (x := self.__it.unwrap_unchecked().next()).is_some():
                    x = x.unwrap_unchecked()
                    buf.popleft()
                    buf.append(x)
                else:
                    self.__buffer = Option.none()
                    self.__it = Option.none()

    def next(self) -> Option[R]:
        self.__push_window()
        return self.__buffer.map(lambda buf: self.__func(buf))

    @te.override
    def fuse(self) -> "MapWindows[T, R]": # type: ignore[override]
        warnings.warn("MapWindows is already fused.", Warning)
        return self

monad_std.iter.impl.once.OnceWith

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/once.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class OnceWith(IterMeta[T], t.Generic[T]):
    __func: Option[t.Callable[[], T]]

    def __init__(self, func: t.Callable[[], T]):
        self.__func = Option.some(func)

    def next(self) -> Option[T]:
        if self.__func.is_some():
            val = self.__func.map(lambda s: s())
            self.__func = Option.none()
            return val
        else:
            return Option.none()

    def nth(self, n: int = 1) -> Option[T]:
        if self.__func.is_none():
            pass
        elif n > 0:
            self.__func = Option.none()
        else:
            val = self.__func.map(lambda s: s())
            self.__func = Option.none()
            return val

        return Option.none()

    def next_chunk(self, n: int = 2) -> Result[t.List[T], t.List[T]]:
        assert n > 0, "Chunk size must be positive"
        if n > 1:
            vale = list(self.__func.map(lambda s: s()).to_iter())
            self.__func = Option.none()
            return Result.of_err(vale)
        else:
            valo: Result[t.List[T], t.List[T]] = self.__func.map(lambda s: [s()]).ok_or([])
            self.__func = Option.none()
            return valo

    def advance_by(self, n: int = 0) -> Result[None, int]:
        if n == 0:
            return Result.of_ok(None)
        elif self.__func.is_none() and n > 0:
            return Result.of_err(n)
        else:
            self.__func = Option.none()
            if n == 1:
                return Result.of_ok(None)
            return Result.of_err(n - 1)

    def fuse(self) -> "OnceWith[T]":  # type: ignore[override]
        warnings.warn("Fusing repeated iterator is meaningless.", Warning)
        return self

    def skip(self, n: int) -> "OnceWith[T]":  # type: ignore[override]
        warnings.warn("Skip repeated iterator is meaningless.", Warning)
        return self

monad_std.iter.impl.partition.PartitionBy

Bases: Generic[T, L, R]

Source code in monad_std/iter/impl/partition.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class PartitionBy(t.Generic[T, L, R]):
    __spliter: t.Callable[[T], Either[L, R]]
    __it: IterMeta[T]
    __end: bool

    __child_1: "PartitionGroup[T, L, R, L]"
    __child_2: "PartitionGroup[T, L, R, R]"

    def __init__(self, __it: IterMeta[T], __spliter: t.Callable[[T], Either[L, R]]):
        self.__spliter = __spliter
        self.__it = __it
        self.__end = False

        self.__child_1 = PartitionGroup(self, lambda: self.__next_left())
        self.__child_2 = PartitionGroup(self, lambda: self.__next_right())

    @staticmethod
    def init(
            __it: IterMeta[T],
            __spliter: t.Callable[[T], Either[L, R]]
    ) -> "t.Tuple[PartitionBy[T, L, R], PartitionGroup[T, L, R, L], PartitionGroup[T, L, R, R]]":
        pb = PartitionBy(__it, __spliter)
        return pb, pb.__child_1, pb.__child_2

    def __next(self) -> Option[Either[L, R]]:
        if self.__end:
            return Option.none()
        nxt = self.__it.next()
        if nxt.is_none():
            self.__end = True
            del self.__it
            return Option.none()
        else:
            return nxt.map(lambda x: self.__spliter(x))

    def __next_left(self) -> Option[L]:
        while True:
            _nxt = self.__next()
            if _nxt.is_none():
                del self.__child_1
                return Option.none()
            nxt = _nxt.unwrap_unchecked()
            if nxt.is_left():
                return Option.some(nxt.unwrap_left_unchecked())
            else:
                # noinspection PyProtectedMember
                self.__child_2._push_buffer(nxt.unwrap_right_unchecked())

    def __next_right(self) -> Option[R]:
        while True:
            _nxt = self.__next()
            if _nxt.is_none():
                del self.__child_2
                return Option.none()
            nxt = _nxt.unwrap_unchecked()
            if nxt.is_right():
                return Option.some(nxt.unwrap_right_unchecked())
            else:
                # noinspection PyProtectedMember
                self.__child_1._push_buffer(nxt.unwrap_left_unchecked())

monad_std.iter.impl.partition.PartitionGroup

Bases: IterMeta[B], Generic[T, L, R, B]

Source code in monad_std/iter/impl/partition.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class PartitionGroup(IterMeta[B], t.Generic[T, L, R, B]):
    __parent: PartitionBy[T, L, R]
    __end: bool
    __buffer: t.Deque[B]
    __parent_next: t.Callable[[], Option[B]]

    def __init__(self, __parent: PartitionBy[T, L, R], __parent_next: t.Callable[[], Option[B]]):
        self.__parent = __parent
        self.__end = False
        self.__parent_next = __parent_next
        self.__buffer = collections.deque()

    def next(self) -> Option[B]:
        if self.__end and self.__buffer:
            return Option.none()
        if len(self.__buffer) == 0:
            nxt = self.__parent_next()
            if nxt.is_none():
                self.__end = True
                del self.__parent, self.__parent_next, self.__buffer
                return Option.none()
            return nxt
        else:
            return Option.some(self.__buffer.popleft())

    def _push_buffer(self, item: B):
        self.__buffer.append(item)

monad_std.iter.impl.peekable.Peekable

Bases: IterMeta[T], Generic[T, It]

Source code in monad_std/iter/impl/peekable.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Peekable(IterMeta[T], t.Generic[T, It]):
    __it: It
    __peek: Option[Option[T]]

    def __init__(self, it: It):
        self.__it = it
        self.__peek = Option.none()

    def next(self) -> Option[T]:
        pk = self.__peek
        self.__peek = Option.none()
        return pk.unwrap_or_else(lambda: self.__it.next())

    def __peek_next(self) -> Option[T]:
        pk = self.__it.next()
        self.__peek = Option.some(pk)
        return pk

    def peek(self) -> Option[T]:
        """Peek the next element of the inner iterator.

        Note that the underlying iterator is still advanced when `peek` is called for the first time:
        In order to retrieve the next element, `next` is called on the underlying iterator,
        hence any side effects (i.e. anything other than fetching the next value) of the next method will occur.

        Examples:
            ```python
            it = siter([1, 2, 3]).peekable()
            assert it.peek() == Option.some(1)
            assert it.next() == Option.some(1)
            assert it.peek() == Option.some(2)
            assert it.peek() == Option.some(2)
            assert it.next() == Option.some(2)
            assert it.next() == Option.some(3)
            assert it.peek() == Option.none()
            assert it.next() == Option.none()
            ```
        """
        return self.__peek.unwrap_or_else(lambda: self.__peek_next())

peek()

Peek the next element of the inner iterator.

Note that the underlying iterator is still advanced when peek is called for the first time: In order to retrieve the next element, next is called on the underlying iterator, hence any side effects (i.e. anything other than fetching the next value) of the next method will occur.

Examples:

1
2
3
4
5
6
7
8
9
it = siter([1, 2, 3]).peekable()
assert it.peek() == Option.some(1)
assert it.next() == Option.some(1)
assert it.peek() == Option.some(2)
assert it.peek() == Option.some(2)
assert it.next() == Option.some(2)
assert it.next() == Option.some(3)
assert it.peek() == Option.none()
assert it.next() == Option.none()
Source code in monad_std/iter/impl/peekable.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def peek(self) -> Option[T]:
    """Peek the next element of the inner iterator.

    Note that the underlying iterator is still advanced when `peek` is called for the first time:
    In order to retrieve the next element, `next` is called on the underlying iterator,
    hence any side effects (i.e. anything other than fetching the next value) of the next method will occur.

    Examples:
        ```python
        it = siter([1, 2, 3]).peekable()
        assert it.peek() == Option.some(1)
        assert it.next() == Option.some(1)
        assert it.peek() == Option.some(2)
        assert it.peek() == Option.some(2)
        assert it.next() == Option.some(2)
        assert it.next() == Option.some(3)
        assert it.peek() == Option.none()
        assert it.next() == Option.none()
        ```
    """
    return self.__peek.unwrap_or_else(lambda: self.__peek_next())

monad_std.iter.impl.repeat.Repeat

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/repeat.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class Repeat(IterMeta[T], t.Generic[T]):
    __val: T

    def __init__(self, value: T):
        self.__val = value

    def next(self) -> Option[T]:
        return Option.some(copy.deepcopy(self.__val))

    def nth(self, n: int = 1) -> Option[T]:
        return Option.some(copy.deepcopy(self.__val))

    def advance_by(self, n: int = 0) -> Result[None, int]:
        return Ok(None)

    def next_chunk(self, n: int = 2) -> Result[t.List[T], t.List[T]]:
        assert n > 0, "Chunk size must be positive"
        return Ok([copy.deepcopy(self.__val) for _ in range(n)])

    def any(self, func: t.Callable[[T], bool] = lambda x: bool(x)) -> bool:
        return func(self.__val)

    def all(self, func: t.Callable[[T], bool] = lambda x: bool(x)) -> bool:
        return func(self.__val)

    def count(self) -> int:
        raise ValueError("Repeat iterator is infinitive and you cannot count it.")

    def find(self, predicate: t.Callable[[T], bool]) -> Option[T]:
        if predicate(self.__val):
            return Option.some(copy.deepcopy(self.__val))
        else:
            return Option.none()

    def find_map(self, func: t.Callable[[T], Option[U]]) -> Option[U]:
        return func(self.__val)

    def fuse(self) -> "Repeat[T]":  # type: ignore[override]
        warnings.warn("Fusing repeated iterator is meaningless.", Warning)
        return self

    def skip(self, n: int) -> "Repeat[T]":  # type: ignore[override]
        warnings.warn("Skip repeated iterator is meaningless.", Warning)
        return self

monad_std.iter.impl.scan.Scan

Bases: IterMeta[B], Generic[T, B, U]

Source code in monad_std/iter/impl/scan.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class Scan(IterMeta[B], t.Generic[T, B, U]):
    __it: IterMeta[T]
    __func: t.Callable[[U, T], t.Tuple[U, Option[B]]]
    __state: U

    def __init__(self, it: IterMeta[T], init: U, func: t.Callable[[U, T], t.Tuple[U, Option[B]]]):
        self.__it = it
        self.__func = func
        self.__state = init

    def __update_state(self, x: T) -> Option[B]:
        st, opt = self.__func(self.__state, x)
        self.__state = st
        return opt

    def next(self) -> Option[B]:
        return self.__it.next().and_then(lambda x: self.__update_state(x))

monad_std.iter.impl.skip.Skip

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/skip.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class Skip(IterMeta[T], t.Generic[T]):
    __skipped: bool
    __skip_n: int
    __it: IterMeta[T]

    def __init__(self, it: IterMeta[T], n: int):
        self.__skipped = False
        self.__skip_n = n
        self.__it = it

    def next(self) -> Option[T]:
        if self.__skipped:
            return self.__it.next()
        else:
            for _ in range(self.__skip_n):
                if self.__it.next().is_none():
                    self.__skipped = True
                    return Option.none()
            self.__skipped = True
            return self.__it.next()

monad_std.iter.impl.take.Take

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/take.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class Take(IterMeta[T], t.Generic[T]):
    __remain: int
    __it: IterMeta[T]

    def __init__(self, it: IterMeta[T], take: int):
        self.__remain = take
        self.__it = it

    def next(self) -> Option[T]:
        if self.__remain != 0:
            self.__remain -= 1
            return self.__it.next()
        else:
            return Option.none()

monad_std.iter.impl.take.TakeWhile

Bases: IterMeta[T], Generic[T]

Source code in monad_std/iter/impl/take.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class TakeWhile(IterMeta[T], t.Generic[T]):
    __func: t.Callable[[T], bool]
    __flag: bool
    __it: IterMeta[T]

    def __init__(self, it: IterMeta[T], func: t.Callable[[T], bool]):
        self.__func = func
        self.__it = it
        self.__flag = False

    def __while_check(self, value: T) -> Option[T]:
        if self.__func(value):
            return Option.some(value)
        else:
            self.__flag = True
            return Option.none()

    def next(self) -> Option[T]:
        if self.__flag:
            return Option.none()
        else:
            return self.__it.next().and_then(lambda x: self.__while_check(x))

monad_std.iter.impl.zip.Zip

Bases: IterMeta[Tuple[T, U]], Generic[T, U, It1, It2]

Source code in monad_std/iter/impl/zip.py 
12
13
14
15
16
17
18
19
20
21
class Zip(IterMeta[t.Tuple[T, U]], t.Generic[T, U, It1, It2]):
    __it1: It1
    __it2: It2

    def __init__(self, iter1: It1, iter2: It2):
        self.__it1 = iter1
        self.__it2 = iter2

    def next(self) -> Option[t.Tuple[T, U]]:
        return self.__it1.next().zip(self.__it2.next())