scaffold.data.iterstream ======================== .. py:module:: scaffold.data.iterstream .. autoapi-nested-parse:: At a low level, iterstream is defined through a class called :py:class:`Composable`. All chaining methods are defined within this class, and can be applied at its high level twin class called :py:class:`IterableSource`. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/scaffold/data/iterstream/base/index /autoapi/scaffold/data/iterstream/iterators/index /autoapi/scaffold/data/iterstream/source/index Classes ------- .. autoapisummary:: scaffold.data.iterstream.Composable scaffold.data.iterstream.FilePathGenerator scaffold.data.iterstream.IterableSamplerSource scaffold.data.iterstream.IterableSource Package Contents ---------------- .. py:class:: Composable(source: Optional[Union[Iterable, Callable]] = None) Bases: :py:obj:`Iterable` A mix-in class that provides stream manipulation functionalities. Init .. py:method:: __iter__() -> Iterator :abstractmethod: Abstract iter .. py:method:: async_map(callback: Callable, buffer: int = 100, max_workers: Optional[int] = None, executor: Optional[concurrent.futures.Executor] = None, **kw) -> _AsyncMap Applies the `callback` to the item in the self and returns the result. :param callback: a callable to be applied to items in the stream :type callback: Callable :param buffer: the size of the buffer :type buffer: int :param max_workers: number of workers in the :py:class:`ThreadPoolExecutor `. `max_workers` is only used when `executor` is not provided, as the `executor` already includes the number of `max_workers`. :type max_workers: int :param executor: an optional executor to be used. By default a :py:class:`ThreadPoolExecutor ` is created, if no executor is provided. If you need a :py:class:`ProcessPoolExecutor `, you can explicitly provide it here. It is also useful when chaining multiple `async_map`; you can pass the same `executor` to each `async_map` to share resources. If `dask.distributed.Client` is passed, tasks will be executed with the provided client (local or remote). Note: if the executor is provided, it will not be closed in this function even after the iterator is exhausted. Note: if executor is provided, the argument ``max_workers`` will be ignored. You should specify this in the executor that is being passed. :type executor: concurrent.futures.Executor, dask.distributed.Client :param \*\*kw: key-word arguments for callback :type \*\*kw: dict Returns (_AsyncMap) .. py:method:: batched(batchsize: int, collation_fn: Optional[Callable] = None, drop_last_if_not_full: bool = True) -> _Iterable Batch items in the stream. :param batchsize: number of items to be batched together :param collation_fn: Collation function to use. :param drop_last_if_not_full: if the length of the last batch is less than the `batchsize`, drop it :type drop_last_if_not_full: bool .. py:method:: collect() -> List[Any] Collect and returns the result of the stream .. py:method:: compose(constructor: Type[Composable], *args, **kw) -> Composable Apply the transformation expressed in the `__iter__` method of the `constructor` to items in the stream. If the provided constructor has an __init__ method, then the source argument should not be provided. .. py:method:: filter(predicate: Callable) -> _Iterable Filters items by `predicate` callable .. py:method:: flatten() -> _Iterable When items in the stream are themselves iterables, flatten turn them back to individual items again .. py:method:: join() -> None A method to consume the stream .. py:method:: loop(n: Optional[int] = None) -> Composable Repeat the iterable n times. :param n: number of times that the iterable is looped over. If None (the default), it loops forever :type n: int, Optional Note: this method creates a deepcopy of the `source` attribute, i.e. all steps in the chain of Composables `before` the loop itself, which must be picklable. .. py:method:: map(callback: Callable, **kw) -> _Iterable Applies the ``callback`` to each item in the stream. Specify key-word arguments for callback in ``kw`` .. py:method:: shuffle(size: Optional[int] = 1000, **kw) -> Composable Shuffles items in the buffer, defined by `size`, to simulate IID sample retrieval. :param size: Buffer size for shuffling. Defaults to 1000. Skip the shuffle step if `size < 2`. :type size: int, optional Acceptable keyword arguments: - initial (int, optional): Minimum number of elements in the buffer before yielding the first element. Must be less than or equal to `size`, otherwise will be set to `size`. Defaults to 100. - rng (random.Random, optional): Either `random` module or a :py:class:`random.Random` instance. If None, a `random.Random()` is used. - seed (Union[int, float, str, bytes, bytearray, None]): A data input that can be used for `random.seed()`. .. py:method:: sliding(window_size: int, *, deepcopy: bool, stride: int = 1, drop_last_if_not_full: bool = True, min_window_size: int = 1, fill_nan_on_partial: bool = False) -> Composable Apply sliding window over the stream. :param window_size: the length of the window :type window_size: int :param deepcopy: If True, each window will be returned as a deepcopy. If items are mutated in the subsequent steps of the pipeline, this should be set to True, otherwise it should be False. Note that deepcopy may incur a substantial cost, so set this parameter carefully. :type deepcopy: bool :param stride: the distance that the window moves at each step :type stride: int :param drop_last_if_not_full: If True, it would only return windows of size `window_size` and drops the last items which have fewer items. :type drop_last_if_not_full: bool :param min_window_size: The minimum length of the window for the last remaining elements. This argument is only relevant if `drop_last_if_not_full` is set to False, otherwise it's ignored. :type min_window_size: int :param fill_nan_on_partial: If `drop_last_if_not_full` is False, the length of the last few windows will be less than `window_size`. This argument fill the missing values with None if set to True. This argument take precedence over If `min_window_size`. :type fill_nan_on_partial: bool .. py:method:: source_(source: Union[Iterable, Callable]) -> Composable Set the source of the stream .. py:method:: take(n: Optional[int]) -> Composable Take n samples from iterable .. py:method:: to(f: Callable, *args, **kw) -> _Iterable Pipe the iterable into another iterable which applies `f` callable on it .. py:method:: tqdm(**kw) -> _Iterable Add tqdm to iterator. .. py:method:: zip_index(pad_length: int = None) -> Composable Zip the item in the stream with its index and yield Tuple[index, item] :param pad_length: if provided, all indexes will be padded with zeros if they have less digits than pad_length, in which case all indexes are str rather than int. .. py:attribute:: source :value: None .. py:class:: FilePathGenerator(url: str, nested: bool = False, max_workers: Optional[int] = None, max_keys: int = 1000000, max_dirs: int = 10, **storage_options) Bases: :py:obj:`scaffold.data.iterstream.base.Composable` A specialized version of `Composable` that accepts a url without instantiating a filesystem instance in the init. It simply generates directories under the given `url` by instantiating a fsspec filesystem and yielding the result of fs.ls(url). :param url: the url for which, ls is performed :param nested: if True, it attempts to make ls on each directory that it encounters. Otherwise, it will only yields the top-level paths and will not expand if the path is a directory :param max_workers: passed to the ThreadPoolExecutor. Only applicable if nested==True :type max_workers: int :param max_keys: maximum number of keys to keep in memory at the same time. If this number is reached, no new expansion on the currently discovered directories is done, until enough keys are yielded to make room for the new ones. :type max_keys: int :param max_dirs: maximum number of parallel ls operation. :type max_dirs: int :param \*\*storage_options: kwargs to pass onto the fsspec filesystem initialization. :type \*\*storage_options: dict .. py:method:: __iter__() -> Iterator[str] Iterator that does ls and yield filepaths under the given url .. py:attribute:: max_dirs :value: 10 .. py:attribute:: max_keys :value: 1000000 .. py:attribute:: max_workers :value: None .. py:attribute:: nested :value: False .. py:attribute:: protocol .. py:attribute:: storage_options .. py:attribute:: url .. py:class:: IterableSamplerSource(iterables: List[Iterable], probs: Optional[List[float]] = None, rng: Optional[random.Random] = None, seed: Optional[int] = None) Bases: :py:obj:`scaffold.data.iterstream.base.Composable` A class that samples from iterables into an iterstream. Initialize IterableSamplerSource. :param iterables: List of iterables to sample from. :type iterables: List[Iterable] :param probs: [description]. Defaults to None. :type probs: Optional[List[float]], optional :param rng: Random number generator to use. :type rng: random.Random, optional :param seed: An int or other acceptable types that works for random.seed(). Will be used to seed `rng`. If None, a unique identifier will be used to seed. :type seed: Optional[int] .. py:method:: __iter__() -> Iterator Samples items from the iterables, returns all samples until all iterables are exhausted. .. py:attribute:: iterables .. py:attribute:: probs :value: None .. py:attribute:: rng :value: None .. py:class:: IterableSource(source: Optional[Union[Iterable, Callable]] = ()) Bases: :py:obj:`scaffold.data.iterstream.base.Composable` A class that turns an iterable to a source of a stream and provides stream manipulation functionalities on top, for instance: - map - map_async - filter - batched - shuffle - and more For the detailed description of each, please refer to the corresponding docstring in :py:class:`Composable`. Initialize IterableSource. :param source: An Iterable that the IterableSource is built based on, or a callable that generates items when called. :type source: Union[Iterable, Callable], Optional .. py:method:: __iter__() -> Iterator Iterates over the items in the iterable