class_docstr: The streaming dataset can be used once your data have been optimised using the DatasetOptimiser class.
class_docstr_tok: ["The","streaming","dataset","can","be","used","once","your","data","have","been","optimised","using","the","DatasetOptimiser","class","."]
code: def _resume(
      self, chunks_replica: List[int], intervals_replica: List[Any]
  ) -> None:
      assert self._state_dict
      assert self.worker_env
      assert self.shuffler
  
      restart_keys = sorted(self._state_dict)
  
      # Get the state from the previous run
      state: Dict[str, Any] = self._state_dict[restart_keys[-1]]
  
      num_workers = state["num_workers"]
      batch_size = state["batch_size"]
  
      # TODO: Implement elastic sampling where the number of workers, ranks can change.
      num_samples_yielded = sum(
          [
              state["num_samples_yielded"]
              for state in self._state_dict.values()
          ]
      )
  
      # replay sampling from each worker / chunks using the batch size
      workers_chunks, workers_intervals = _associate_chunks_to_workers(
          num_workers,
          self.worker_env,
          chunks_replica,
          intervals_replica,
      )
      indexes = _replay_sampling(
          num_samples_yielded, batch_size, num_workers
      )
      chunks_index, indexes = _replay_chunks_sampling(
          workers_intervals, indexes
      )
  
      # select the chunks and intervals associated to this worker
      worker_rank = self.worker_env.rank
      self.num_chunks = len(workers_intervals[worker_rank])
      self.chunk_index = chunks_index[worker_rank]
      self.worker_chunks = workers_chunks[worker_rank]
      self.worker_intervals = workers_intervals[worker_rank]
  
      # replay the indexes for the current chunks
      interval = workers_intervals[worker_rank][self.chunk_index]
      current_indexes = np.arange(interval[0], interval[1])
  
      # re-shuffle the indexes
      current_indexes = self.shuffler(
          current_indexes,
          self.num_chunks,
          self.current_epoch,
          self.chunk_index,
      )
  
      # skip any indexes already consumed
      current_indexes = current_indexes[indexes[worker_rank] :]
      self.current_indexes = current_indexes
  
      # bump the chunk_index
      self.chunk_index += 1
  
imports: import hashlib
  import os
  from dataclasses import dataclass
  from time import time
  from typing import Any, Dict, List, Optional, Tuple, Union
  import numpy
  from torch.utils.data import IterableDataset, get_worker_info
  from lightning.data.streaming import Cache
  from lightning.data.streaming.constants import _DEFAULT_CACHE_DIR, _INDEX_FILENAME, _LIGHTNING_CLOUD_LATEST
  from lightning.data.streaming.item_loader import BaseItemLoader
  from lightning.data.streaming.sampler import ChunkedIndex
  from lightning.data.streaming.serializers import Serializer
  from lightning.data.streaming.shuffle import FullShuffle, NoShuffle, Shuffle
  from lightning.data.utilities.env import Environment, _DistributedEnv, _WorkerEnv
label_desc: Use image node_id 6 for calling the StreamingDataset obj's underlying member method code with example usage: obj._resume(chunks_replica, intervals_replica) without return types