Skip to content

bin_packing

PackingIterator

Bases: Generic[T]

Source code in pyiceberg/utils/bin_packing.py 
 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 PackingIterator(Generic[T]):
    bins: List[Bin[T]]

    def __init__(
        self,
        items: Iterable[T],
        target_weight: int,
        lookback: int,
        weight_func: Callable[[T], int],
        largest_bin_first: bool = False,
    ) -> None:
        self.items = iter(items)
        self.target_weight = target_weight
        self.lookback = lookback
        self.weight_func = weight_func
        self.largest_bin_first = largest_bin_first
        self.bins = []

    def __iter__(self) -> PackingIterator[T]:
        """Return an iterator for the PackingIterator class."""
        return self

    def __next__(self) -> List[T]:
        """Return the next item when iterating over the PackingIterator class."""
        while True:
            try:
                item = next(self.items)
                weight = self.weight_func(item)
                bin_ = self.find_bin(weight)
                if bin_ is not None:
                    bin_.add(item, weight)
                else:
                    bin_ = Bin(self.target_weight)
                    bin_.add(item, weight)
                    self.bins.append(bin_)

                    if len(self.bins) > self.lookback:
                        return self.remove_bin().items
            except StopIteration:
                break

        if len(self.bins) == 0:
            raise StopIteration()

        return self.remove_bin().items

    def find_bin(self, weight: int) -> Optional[Bin[T]]:
        for bin_ in self.bins:
            if bin_.can_add(weight):
                return bin_
        return None

    def remove_bin(self) -> Bin[T]:
        if self.largest_bin_first:
            bin_ = max(self.bins, key=lambda b: b.weight())
            self.bins.remove(bin_)
            return bin_
        else:
            return self.bins.pop(0)

__iter__()

Return an iterator for the PackingIterator class.

Source code in pyiceberg/utils/bin_packing.py 
66
67
68
def __iter__(self) -> PackingIterator[T]:
    """Return an iterator for the PackingIterator class."""
    return self

__next__()

Return the next item when iterating over the PackingIterator class.

Source code in pyiceberg/utils/bin_packing.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __next__(self) -> List[T]:
    """Return the next item when iterating over the PackingIterator class."""
    while True:
        try:
            item = next(self.items)
            weight = self.weight_func(item)
            bin_ = self.find_bin(weight)
            if bin_ is not None:
                bin_.add(item, weight)
            else:
                bin_ = Bin(self.target_weight)
                bin_.add(item, weight)
                self.bins.append(bin_)

                if len(self.bins) > self.lookback:
                    return self.remove_bin().items
        except StopIteration:
            break

    if len(self.bins) == 0:
        raise StopIteration()

    return self.remove_bin().items