scaffold.data.iterstream.base ============================= .. py:module:: scaffold.data.iterstream.base Classes ------- .. autoapisummary:: scaffold.data.iterstream.base.AsyncContent scaffold.data.iterstream.base.Composable Module Contents --------------- .. py:class:: AsyncContent(item: str, func: Callable, executor: concurrent.futures.Executor) Represents content that can be fetched asynchronously. Initialize AsyncContent. :param item: Key corresponding to a single item, will be passed to `fetch_func`. :type item: str :param func: Function that fetches a given key. :type func: Callable :param executor: Executor to submit `func` with `item`. :type executor: concurrent.futures.Executor .. py:method:: value(timeout: int = None) -> Any Get the value asynchronously. :param timeout: Number of seconds to wait for the result. If None, then the future is waited indefinitely. Defaults to None. :type timeout: int, optional :returns: Content. :rtype: Any .. py:attribute:: future .. py:attribute:: stack :value: 1 .. 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