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 __iter__(self) -> "StreamingDataset":
      self.worker_env = _WorkerEnv.detect()
      self.cache = self._create_cache(worker_env=self.worker_env)
      self.shuffler = self._create_shuffler(self.cache)
  
      # Handle restart
      if self._state_dict:
          self._validate_state_dict()
          restart_keys = sorted(self._state_dict)
          state: Dict[str, Any] = self._state_dict[restart_keys[-1]]
          self.current_epoch = state["current_epoch"]
  
      (
          chunks_per_replica,
          intervals_per_replica,
      ) = self.shuffler.get_chunks_and_intervals_per_ranks(
          self.distributed_env, self.current_epoch
      )
      chunks_replica = chunks_per_replica[
          self.distributed_env.global_rank
          % self.distributed_env.world_size
      ]
      intervals_replica = intervals_per_replica[
          self.distributed_env.global_rank
          % self.distributed_env.world_size
      ]
  
      # Handle restart
      if self._state_dict:
          self._resume(chunks_replica, intervals_replica)
      else:
          (
              chunks_per_replica,
              intervals_per_replica,
          ) = self.shuffler.get_chunks_and_intervals_per_ranks(
              self.distributed_env, self.current_epoch
          )
          chunks_replica = chunks_per_replica[
              self.distributed_env.global_rank
              % self.distributed_env.world_size
          ]
          intervals_replica = intervals_per_replica[
              self.distributed_env.global_rank
              % self.distributed_env.world_size
          ]
  
          self.worker_chunks = []
          self.worker_intervals = []
  
          for i, (chunk_index, chunk_interval) in enumerate(
              zip(chunks_replica, intervals_replica)
          ):
              if i % self.worker_env.world_size != self.worker_env.rank:
                  continue
              self.worker_chunks.append(chunk_index)
              self.worker_intervals.append(chunk_interval)
  
          self.num_chunks = len(self.worker_chunks)
  
          self.current_indexes = []
          self.chunk_index = 0
          self.global_index = 0
          self.index = 0
  
      self.has_triggered_download = False
      self.last_time = time()
  
      return self
  
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 5 for calling the StreamingDataset obj's underlying member method code with example usage: obj.__iter__() and returns: self