python/ray/data/dataset_pipeline.py
def repeat(self, times: int = None) -> "DatasetPipeline[T]":
        """Repeat this pipeline a given number or times, or indefinitely.

        This operation is only allowed for pipelines of a finite length. An
        error will be raised for pipelines of infinite length.

        Note that every repeat of the pipeline is considered an "epoch" for
        the purposes of ``iter_epochs()``. If there are multiple repeat calls,
        the latest repeat takes precedence for the purpose of defining epochs.

        Args:
            times: The number of times to loop over this pipeline, or None
                to repeat indefinitely.
        """

        if self._length == float("inf"):
            raise ValueError("Cannot repeat a pipeline of infinite length.")

        class RepeatIterator:
            def __init__(self, original_iter):
                self._original_iter = original_iter
                # Holds results to repeat.
                self._results = []
                # Incrementing cursor over results.
                self._i = 0
                # This is calculated later.
                self._max_i = None

            def __next__(self) -> Dataset[T]:
                # Still going through the original pipeline.
                if self._original_iter:
                    try:
                        make_ds = next(self._original_iter)
                        self._results.append(make_ds)

                        def gen():
                            res = make_ds()
                            res._set_epoch(0)
                            return res

                        return gen
                    except StopIteration:
                        self._original_iter = None
                        # Calculate the cursor limit.
                        if times:
                            self._max_i = len(self._results) * (times - 1)
                        else:
                            self._max_i = float("inf")
                # Going through a repeat of the pipeline.
                if self._i < self._max_i:
                    make_ds = self._results[self._i % len(self._results)]
                    epoch = 1 + self._i // len(self._results)

                    def gen():
                        res = make_ds()
                        res._set_epoch(epoch)
                        return res

                    self._i += 1
                    return gen
                else:
                    raise StopIteration

        class RepeatIterable:
            def __init__(self, original_iter):
                self._original_iter = original_iter

            def __iter__(self):
                return RepeatIterator(self._original_iter)

        if not times:
            length = float("inf")
        elif times and self._length:
            length = times * self._length
        else:
            length = None

        return DatasetPipeline(
            RepeatIterable(iter(self._base_iterable)),
            stages=self._stages.copy(),
            length=length,
        )
Repeat this pipeline a given number or times, or indefinitely.

        This operation is only allowed for pipelines of a finite length. An
        error will be raised for pipelines of infinite length.

        Note that every repeat of the pipeline is considered an "epoch" for
        the purposes of ``iter_epochs()``. If there are multiple repeat calls,
        the latest repeat takes precedence for the purpose of defining epochs.

        Args:
            times: The number of times to loop over this pipeline, or None
                to repeat indefinitely.
        
Repeat this pipeline a given number or times, or indefinitely.
This operation is only allowed for pipelines of a finite length. An
error will be raised for pipelines of infinite length.

Note that every repeat of the pipeline is considered an "epoch" for
the purposes of ``iter_epochs()``. If there are multiple repeat calls,
the latest repeat takes precedence for the purpose of defining epochs.
Repeat this pipeline a given number or times or indefinitely . This operation is only allowed for pipelines of a finite length . An error will be raised for pipelines of infinite length . Note that every repeat of the pipeline is considered an " epoch " for the purposes of ` ` iter_epochs () ` ` . If there are multiple repeat calls the latest repeat takes precedence for the purpose of defining epochs .
def repeat(self, times: int = None) -> "DatasetPipeline[T]":
        if self._length == float("inf"):
            raise ValueError("Cannot repeat a pipeline of infinite length.")
        class RepeatIterator:
            def __init__(self, original_iter):
                self._original_iter = original_iter
                self._results = []
                self._i = 0
                self._max_i = None
            def __next__(self) -> Dataset[T]:
                if self._original_iter:
                    try:
                        make_ds = next(self._original_iter)
                        self._results.append(make_ds)
                        def gen():
                            res = make_ds()
                            res._set_epoch(0)
                            return res
                        return gen
                    except StopIteration:
                        self._original_iter = None
                        if times:
                            self._max_i = len(self._results) * (times - 1)
                        else:
                            self._max_i = float("inf")
                if self._i < self._max_i:
                    make_ds = self._results[self._i % len(self._results)]
                    epoch = 1 + self._i // len(self._results)
                    def gen():
                        res = make_ds()
                        res._set_epoch(epoch)
                        return res
                    self._i += 1
                    return gen
                else:
                    raise StopIteration
        class RepeatIterable:
            def __init__(self, original_iter):
                self._original_iter = original_iter
            def __iter__(self):
                return RepeatIterator(self._original_iter)
        if not times:
            length = float("inf")
        elif times and self._length:
            length = times * self._length
        else:
            length = None
        return DatasetPipeline(
            RepeatIterable(iter(self._base_iterable)),
            stages=self._stages.copy(),
            length=length,
        )
def repeat ( self , times : int = None ) -> "DatasetPipeline[T]" : if self . _length == float ( "inf" ) : raise ValueError ( "Cannot repeat a pipeline of infinite length." ) class RepeatIterator : def __init__ ( self , original_iter ) : self . _original_iter = original_iter self . _results = [ ] self . _i = 0 self . _max_i = None def __next__ ( self ) -> Dataset [ T ] : if self . _original_iter : try : make_ds = next ( self . _original_iter ) self . _results . append ( make_ds ) def gen ( ) : res = make_ds ( ) res . _set_epoch ( 0 ) return res return gen except StopIteration : self . _original_iter = None if times : self . _max_i = len ( self . _results ) * ( times - 1 ) else : self . _max_i = float ( "inf" ) if self . _i < self . _max_i : make_ds = self . _results [ self . _i % len ( self . _results ) ] epoch = 1 + self . _i // len ( self . _results ) def gen ( ) : res = make_ds ( ) res . _set_epoch ( epoch ) return res self . _i += 1 return gen else : raise StopIteration class RepeatIterable : def __init__ ( self , original_iter ) : self . _original_iter = original_iter def __iter__ ( self ) : return RepeatIterator ( self . _original_iter ) if not times : length = float ( "inf" ) elif times and self . _length : length = times * self . _length else : length = None return DatasetPipeline ( RepeatIterable ( iter ( self . _base_iterable ) ) , stages = self . _stages . copy ( ) , length = length , )
Repeat this pipeline a given number or times, or indefinitely.
Repeat this pipeline a given number or times or indefinitely .
"""Repeat this pipeline a given number or times, or indefinitely.

        This operation is only allowed for pipelines of a finite length. An
        error will be raised for pipelines of infinite length.

        Note that every repeat of the pipeline is considered an "epoch" for
        the purposes of ``iter_epochs()``. If there are multiple repeat calls,
        the latest repeat takes precedence for the purpose of defining epochs.

        Args:
            times: The number of times to loop over this pipeline, or None
                to repeat indefinitely.
        """ # Holds results to repeat. # Incrementing cursor over results. # This is calculated later. # Still going through the original pipeline. # Calculate the cursor limit. # Going through a repeat of the pipeline.
self
times
int
The number of times to loop over this pipeline, or None
to repeat indefinitely.
The number of times to loop over this pipeline or None to repeat indefinitely .