Skip to content

Heap

Bases: PyoMutableSequence[T], ABC


              flowchart TD
              pyochain.collections._heap.Heap[Heap]
              pyochain.abc._sequences.PyoMutableSequence[PyoMutableSequence]
              pyochain.abc._sequences.PyoSequence[PyoSequence]
              pyochain.abc._sequences.PyoReversible[PyoReversible]
              pyochain.abc._collection.PyoCollection[PyoCollection]
              pyochain.abc._iterable.PyoIterable[PyoIterable]
              pyochain.abc._collection.PyoContainer[PyoContainer]
              pyochain.abc._collection.PyoSized[PyoSized]
              pyochain.abc._mixins.Checkable[Checkable]
              pyochain.abc._mixins.Fluent[Fluent]
              pyochain.abc._mixins.Pipe[Pipe]
              pyochain.abc._mixins.Tap[Tap]

                              pyochain.abc._sequences.PyoMutableSequence --> pyochain.collections._heap.Heap
                                pyochain.abc._sequences.PyoSequence --> pyochain.abc._sequences.PyoMutableSequence
                                pyochain.abc._sequences.PyoReversible --> pyochain.abc._sequences.PyoSequence
                                pyochain.abc._iterable.PyoIterable --> pyochain.abc._sequences.PyoReversible
                                pyochain.abc._mixins.Checkable --> pyochain.abc._iterable.PyoIterable
                
                pyochain.abc._mixins.Fluent --> pyochain.abc._iterable.PyoIterable
                                pyochain.abc._mixins.Pipe --> pyochain.abc._mixins.Fluent
                
                pyochain.abc._mixins.Tap --> pyochain.abc._mixins.Fluent
                



                pyochain.abc._collection.PyoCollection --> pyochain.abc._sequences.PyoSequence
                                pyochain.abc._iterable.PyoIterable --> pyochain.abc._collection.PyoCollection
                                pyochain.abc._mixins.Checkable --> pyochain.abc._iterable.PyoIterable
                
                pyochain.abc._mixins.Fluent --> pyochain.abc._iterable.PyoIterable
                                pyochain.abc._mixins.Pipe --> pyochain.abc._mixins.Fluent
                
                pyochain.abc._mixins.Tap --> pyochain.abc._mixins.Fluent
                


                pyochain.abc._collection.PyoContainer --> pyochain.abc._collection.PyoCollection
                                pyochain.abc._mixins.Checkable --> pyochain.abc._collection.PyoContainer
                

                pyochain.abc._collection.PyoSized --> pyochain.abc._collection.PyoCollection
                                pyochain.abc._mixins.Checkable --> pyochain.abc._collection.PyoSized
                






              click pyochain.collections._heap.Heap href "" "pyochain.collections._heap.Heap"
              click pyochain.abc._sequences.PyoMutableSequence href "" "pyochain.abc._sequences.PyoMutableSequence"
              click pyochain.abc._sequences.PyoSequence href "" "pyochain.abc._sequences.PyoSequence"
              click pyochain.abc._sequences.PyoReversible href "" "pyochain.abc._sequences.PyoReversible"
              click pyochain.abc._collection.PyoCollection href "" "pyochain.abc._collection.PyoCollection"
              click pyochain.abc._iterable.PyoIterable href "" "pyochain.abc._iterable.PyoIterable"
              click pyochain.abc._collection.PyoContainer href "" "pyochain.abc._collection.PyoContainer"
              click pyochain.abc._collection.PyoSized href "" "pyochain.abc._collection.PyoSized"
              click pyochain.abc._mixins.Checkable href "" "pyochain.abc._mixins.Checkable"
              click pyochain.abc._mixins.Fluent href "" "pyochain.abc._mixins.Fluent"
              click pyochain.abc._mixins.Pipe href "" "pyochain.abc._mixins.Pipe"
              click pyochain.abc._mixins.Tap href "" "pyochain.abc._mixins.Tap"
            

Abstract base class for heaps.

Source code in pyochain/collections/_heap.pyi
 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
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class Heap[T: SupportsRichComparison](PyoMutableSequence[T], ABC):
    """Abstract base class for heaps."""

    def __new__(cls, data: Iterable[T]) -> Self: ...
    @override
    def __len__(self) -> int: ...
    @overload
    def __getitem__(self, i: SupportsIndex, /) -> T: ...
    @overload
    def __getitem__(self, s: slice[SupportsIndex | None], /) -> Vec[T]: ...
    @override
    def __getitem__(
        self, index: SupportsIndex | slice[SupportsIndex | None]
    ) -> T | Vec[T]: ...
    @overload
    def __setitem__(self, index: int, value: T) -> None: ...
    @overload
    def __setitem__(self, index: slice, value: Iterable[T]) -> None: ...
    @override
    def __setitem__(self, index: int | slice, value: T | Iterable[T]) -> None: ...
    @override
    def __delitem__(self, index: int | slice) -> None: ...
    @override
    def __eq__(self, other: object) -> bool: ...
    @staticmethod
    def from_ref[S: SupportsRichComparison](data: list[S]) -> Heap[S]:
        """Create a `Heap` instance from an existing `list` without copying.

        Assumes that the provided list already satisfies the corresponding heap invariant.

        Args:
            data (list[S]): A `list` that already satisfies the corresponding heap invariant.

        Returns:
            Heap[S]: A new `Heap` instance.
        """

    @abstractmethod
    def push(self, item: T) -> None:
        """Push item onto heap, maintaining the heap invariant."""

    @abstractmethod
    def replace(self, item: T) -> T:
        """Pop and return the current smallest value, and add the new item.

        This is more efficient than `pop()` followed by `push()`, and can be
        more appropriate when using a fixed-size heap.

        Note that the value returned may be larger than item!

        That constrains reasonable uses of this routine unless written as part of a conditional replacement:

        Example:
            ```python
            from pyochain.collections import HeapMin

            heap = HeapMin([1, 2, 3])
            item = 4
            item = heap.replace(item) if item > heap[0] else item
            assert heap == HeapMin([2, 4, 3])
            ```

        Args:
            item (T): The new item to be added to the heap.

        Returns:
            T: The smallest item from the heap.
        """

    @abstractmethod
    def push_pop(self, item: T) -> T:
        """Fast version of a heappush followed by a heappop.

        Args:
            item (T): The new item to be added to the heap.

        Returns:
            T: The smallest item from the `Heap`.
        """

    @abstractmethod
    @override
    def pop(self, _index: int = -1, /) -> T:
        """Pop the smallest item off the heap, maintaining the heap invariant.

        Warning:
            *index* is kept to maintain compatibility with the `collections.abc.MutableSequence` interface, but it is ignored.

            The smallest item is always popped.

        Args:
            _index (int): Ignored.

        Returns:
            T: The smallest item from the heap.
        """

    @override
    def insert(self, index: SupportsIndex, value: T) -> None: ...
    def merge[S: SupportsRichComparison](
        self,
        *others: Iterable[S],
        key: Callable[[T | S], SupportsRichComparison] | None = None,
        reverse: bool = False,
    ) -> PyoIterator[T | S]:
        """Merge *self* and *others* into a single sorted output.

        Similar to `vec.iter().chain(*others).sort()`, but:

        - returns an `Iterator`
        - does not pull the data into memory all at once
        - assumes that each of the input streams is already sorted (smallest to largest).

        ```python
        from pyochain.collections import HeapMin

        base = [1, 3, 5, 7]
        x = HeapMin(base).merge([0, 2, 4, 8], [5, 10, 15, 20], [], [25]).collect(list)
        assert x == [0, 1, 2, 3, 4, 5, 5, 7, 8, 10, 15, 20, 25]
        ```

        If *key* is not `None`, applies a key function to each element to determine its sort order.

        ```python
        others = ["cat", "fish", "kangaroo"]
        x = HeapMin(["dog", "horse"]).merge(others, key=len).collect(list)
        assert x == ["dog", "cat", "fish", "horse", "kangaroo"]
        ```

        Args:
            *others (Iterable[S]): Other sorted iterables to merge with *self*.
            key (Callable[[T | S], SupportsRichComparison] | None): A function that extracts a comparison key from each element. Defaults to `None`.
            reverse (bool): If `True`, the merged output will be in descending order. Defaults to `False`.

        Returns:
            PyoIterator[T | S]: A generator that yields the merged sorted elements from the input iterables.

        """

    @overload
    def n_smallest(
        self, n: int, key: Callable[[T], SupportsRichComparison]
    ) -> Self: ...
    @overload
    def n_smallest(self, n: int, key: None = None) -> Self: ...
    def n_smallest(
        self, n: int, key: Callable[[T], SupportsRichComparison] | None = None
    ) -> Self:
        """Find the *n* smallest elements in a dataset.

        Equivalent to:  `iterator.sort(key=key)[:n]`

        Args:
            n (int): The number of smallest elements to retrieve.
            key (Callable[[T], SupportsRichComparison] | None): A function that extracts a comparison key from each element. Defaults to `None`.

        Returns:
            Self: A new `Heap` instance containing the *n* smallest elements from the heap.
        """

    @overload
    def n_largest(self, n: int, key: Callable[[T], SupportsRichComparison]) -> Self: ...
    @overload
    def n_largest(self, n: int, key: None = None) -> Self: ...
    def n_largest(
        self, n: int, key: Callable[[T], SupportsRichComparison] | None = None
    ) -> Self:
        """Find the *n* largest elements in a dataset.

        Equivalent to:  `iterator.sort(key=key, reverse=True)[:n]`

        Args:
            n (int): The number of largest elements to retrieve.
            key (Callable[[T], SupportsRichComparison] | None): A function that extracts a comparison key from each element. Defaults to `None`.

        Returns:
            Self: A new `Heap` instance containing the *n* largest elements from the heap.
        """

from_ref(data) staticmethod

Create a Heap instance from an existing list without copying.

Assumes that the provided list already satisfies the corresponding heap invariant.

Parameters:

Name Type Description Default
data list[S]

A list that already satisfies the corresponding heap invariant.

required

Returns:

Type Description
Heap[S]

Heap[S]: A new Heap instance.

Source code in pyochain/collections/_heap.pyi
34
35
36
37
38
39
40
41
42
43
44
45
@staticmethod
def from_ref[S: SupportsRichComparison](data: list[S]) -> Heap[S]:
    """Create a `Heap` instance from an existing `list` without copying.

    Assumes that the provided list already satisfies the corresponding heap invariant.

    Args:
        data (list[S]): A `list` that already satisfies the corresponding heap invariant.

    Returns:
        Heap[S]: A new `Heap` instance.
    """

push(item) abstractmethod

Push item onto heap, maintaining the heap invariant.

Source code in pyochain/collections/_heap.pyi
47
48
49
@abstractmethod
def push(self, item: T) -> None:
    """Push item onto heap, maintaining the heap invariant."""

replace(item) abstractmethod

Pop and return the current smallest value, and add the new item.

This is more efficient than pop() followed by push(), and can be more appropriate when using a fixed-size heap.

Note that the value returned may be larger than item!

That constrains reasonable uses of this routine unless written as part of a conditional replacement:

Example
from pyochain.collections import HeapMin

heap = HeapMin([1, 2, 3])
item = 4
item = heap.replace(item) if item > heap[0] else item
assert heap == HeapMin([2, 4, 3])

Parameters:

Name Type Description Default
item T

The new item to be added to the heap.

required

Returns:

Name Type Description
T T

The smallest item from the heap.

Source code in pyochain/collections/_heap.pyi
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
@abstractmethod
def replace(self, item: T) -> T:
    """Pop and return the current smallest value, and add the new item.

    This is more efficient than `pop()` followed by `push()`, and can be
    more appropriate when using a fixed-size heap.

    Note that the value returned may be larger than item!

    That constrains reasonable uses of this routine unless written as part of a conditional replacement:

    Example:
        ```python
        from pyochain.collections import HeapMin

        heap = HeapMin([1, 2, 3])
        item = 4
        item = heap.replace(item) if item > heap[0] else item
        assert heap == HeapMin([2, 4, 3])
        ```

    Args:
        item (T): The new item to be added to the heap.

    Returns:
        T: The smallest item from the heap.
    """

push_pop(item) abstractmethod

Fast version of a heappush followed by a heappop.

Parameters:

Name Type Description Default
item T

The new item to be added to the heap.

required

Returns:

Name Type Description
T T

The smallest item from the Heap.

Source code in pyochain/collections/_heap.pyi
79
80
81
82
83
84
85
86
87
88
@abstractmethod
def push_pop(self, item: T) -> T:
    """Fast version of a heappush followed by a heappop.

    Args:
        item (T): The new item to be added to the heap.

    Returns:
        T: The smallest item from the `Heap`.
    """

pop(_index=-1) abstractmethod

Pop the smallest item off the heap, maintaining the heap invariant.

Warning

index is kept to maintain compatibility with the collections.abc.MutableSequence interface, but it is ignored.

The smallest item is always popped.

Parameters:

Name Type Description Default
_index int

Ignored.

-1

Returns:

Name Type Description
T T

The smallest item from the heap.

Source code in pyochain/collections/_heap.pyi
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@abstractmethod
@override
def pop(self, _index: int = -1, /) -> T:
    """Pop the smallest item off the heap, maintaining the heap invariant.

    Warning:
        *index* is kept to maintain compatibility with the `collections.abc.MutableSequence` interface, but it is ignored.

        The smallest item is always popped.

    Args:
        _index (int): Ignored.

    Returns:
        T: The smallest item from the heap.
    """

merge(*others, key=None, reverse=False)

Merge self and others into a single sorted output.

Similar to vec.iter().chain(*others).sort(), but:

  • returns an Iterator
  • does not pull the data into memory all at once
  • assumes that each of the input streams is already sorted (smallest to largest).
from pyochain.collections import HeapMin

base = [1, 3, 5, 7]
x = HeapMin(base).merge([0, 2, 4, 8], [5, 10, 15, 20], [], [25]).collect(list)
assert x == [0, 1, 2, 3, 4, 5, 5, 7, 8, 10, 15, 20, 25]

If key is not None, applies a key function to each element to determine its sort order.

others = ["cat", "fish", "kangaroo"]
x = HeapMin(["dog", "horse"]).merge(others, key=len).collect(list)
assert x == ["dog", "cat", "fish", "horse", "kangaroo"]

Parameters:

Name Type Description Default
*others Iterable[S]

Other sorted iterables to merge with self.

()
key Callable[[T | S], SupportsRichComparison] | None

A function that extracts a comparison key from each element. Defaults to None.

None
reverse bool

If True, the merged output will be in descending order. Defaults to False.

False

Returns:

Type Description
PyoIterator[T | S]

PyoIterator[T | S]: A generator that yields the merged sorted elements from the input iterables.

Source code in pyochain/collections/_heap.pyi
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def merge[S: SupportsRichComparison](
    self,
    *others: Iterable[S],
    key: Callable[[T | S], SupportsRichComparison] | None = None,
    reverse: bool = False,
) -> PyoIterator[T | S]:
    """Merge *self* and *others* into a single sorted output.

    Similar to `vec.iter().chain(*others).sort()`, but:

    - returns an `Iterator`
    - does not pull the data into memory all at once
    - assumes that each of the input streams is already sorted (smallest to largest).

    ```python
    from pyochain.collections import HeapMin

    base = [1, 3, 5, 7]
    x = HeapMin(base).merge([0, 2, 4, 8], [5, 10, 15, 20], [], [25]).collect(list)
    assert x == [0, 1, 2, 3, 4, 5, 5, 7, 8, 10, 15, 20, 25]
    ```

    If *key* is not `None`, applies a key function to each element to determine its sort order.

    ```python
    others = ["cat", "fish", "kangaroo"]
    x = HeapMin(["dog", "horse"]).merge(others, key=len).collect(list)
    assert x == ["dog", "cat", "fish", "horse", "kangaroo"]
    ```

    Args:
        *others (Iterable[S]): Other sorted iterables to merge with *self*.
        key (Callable[[T | S], SupportsRichComparison] | None): A function that extracts a comparison key from each element. Defaults to `None`.
        reverse (bool): If `True`, the merged output will be in descending order. Defaults to `False`.

    Returns:
        PyoIterator[T | S]: A generator that yields the merged sorted elements from the input iterables.

    """

n_smallest(n, key=None)

n_smallest(
    n: int, key: Callable[[T], SupportsRichComparison]
) -> Self
n_smallest(n: int, key: None = None) -> Self

Find the n smallest elements in a dataset.

Equivalent to: iterator.sort(key=key)[:n]

Parameters:

Name Type Description Default
n int

The number of smallest elements to retrieve.

required
key Callable[[T], SupportsRichComparison] | None

A function that extracts a comparison key from each element. Defaults to None.

None

Returns:

Name Type Description
Self Self

A new Heap instance containing the n smallest elements from the heap.

Source code in pyochain/collections/_heap.pyi
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def n_smallest(
    self, n: int, key: Callable[[T], SupportsRichComparison] | None = None
) -> Self:
    """Find the *n* smallest elements in a dataset.

    Equivalent to:  `iterator.sort(key=key)[:n]`

    Args:
        n (int): The number of smallest elements to retrieve.
        key (Callable[[T], SupportsRichComparison] | None): A function that extracts a comparison key from each element. Defaults to `None`.

    Returns:
        Self: A new `Heap` instance containing the *n* smallest elements from the heap.
    """

n_largest(n, key=None)

n_largest(
    n: int, key: Callable[[T], SupportsRichComparison]
) -> Self
n_largest(n: int, key: None = None) -> Self

Find the n largest elements in a dataset.

Equivalent to: iterator.sort(key=key, reverse=True)[:n]

Parameters:

Name Type Description Default
n int

The number of largest elements to retrieve.

required
key Callable[[T], SupportsRichComparison] | None

A function that extracts a comparison key from each element. Defaults to None.

None

Returns:

Name Type Description
Self Self

A new Heap instance containing the n largest elements from the heap.

Source code in pyochain/collections/_heap.pyi
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def n_largest(
    self, n: int, key: Callable[[T], SupportsRichComparison] | None = None
) -> Self:
    """Find the *n* largest elements in a dataset.

    Equivalent to:  `iterator.sort(key=key, reverse=True)[:n]`

    Args:
        n (int): The number of largest elements to retrieve.
        key (Callable[[T], SupportsRichComparison] | None): A function that extracts a comparison key from each element. Defaults to `None`.

    Returns:
        Self: A new `Heap` instance containing the *n* largest elements from the heap.
    """