Skip to content

bin_packing

PackingIterator

Bases: Generic[T]

Source code in pyiceberg/utils/bin_packing.py
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