code: def __init__(
      self,
      data_efficiency_config,
      one_epoch_total_samples,
      micro_batch_size,
      data_parallel_rank,
      data_parallel_size,
      data_parallel_group,
      gradient_accumulation_steps,
      global_rank,
      drop_last=True,
  ):
      # Keep a copy of input params for later use.
      self.data_efficiency_config = data_efficiency_config
      self.one_epoch_total_samples = one_epoch_total_samples
      self.index_dtype = find_fit_int_dtype(0, one_epoch_total_samples)
      self.total_samples = (
          one_epoch_total_samples
          * self.data_efficiency_config[DATA_SAMPLING][
              DATA_SAMPLING_NUM_EPOCHS
          ]
      )
      self.micro_batch_size = micro_batch_size
      self.data_parallel_rank = data_parallel_rank
      self.data_parallel_group = data_parallel_group
      self.micro_batch_times_data_parallel_size = (
          self.micro_batch_size * data_parallel_size
      )
      self.gradient_accumulation_steps = gradient_accumulation_steps
      self.global_batch_size = (
          self.micro_batch_times_data_parallel_size
          * self.gradient_accumulation_steps
      )
      self.global_rank = global_rank
      self.drop_last = drop_last
      self.np_rng = np.random.default_rng(
          self.data_efficiency_config[DATA_EFFICIENCY_SEED]
      )
      self.state = {}
      self.batch = []
      self.consumed_samples = 0
      if self.data_efficiency_config[DATA_SAMPLING][
          CURRICULUM_LEARNING
      ][CURRICULUM_LEARNING_ENABLED]:
          self.curriculum_step = 0
          self.current_difficulties = {}
          self.data_cluster_paths = []
          self.data_cluster_current_position = []
          self.curriculum_schedulers = {}
          self.curriculum_index_to_sample = {}
          self.curriculum_index_to_metric = {}
          self.difficulty_type = {}
          self.clustering_type = {}
          self.data_1epoch_size = None
          if self.global_rank == 0:
              self.data_clusters = []
              self.data_cluster_sizes = []
              cluster_path = self.data_efficiency_config[DATA_SAMPLING][
                  CURRICULUM_LEARNING
              ][CURRICULUM_LEARNING_CLUSTER_PATH]
              if not os.path.exists(cluster_path):
                  os.makedirs(cluster_path)
          for metric in self.data_efficiency_config[DATA_SAMPLING][
              CURRICULUM_LEARNING
          ][CURRICULUM_LEARNING_METRICS]:
              self.curriculum_schedulers[metric] = CurriculumScheduler(
                  data_efficiency_config[DATA_SAMPLING][
                      CURRICULUM_LEARNING
                  ][CURRICULUM_LEARNING_METRICS][metric]
              )
              self.difficulty_type[metric] = data_efficiency_config[
                  DATA_SAMPLING
              ][CURRICULUM_LEARNING][CURRICULUM_LEARNING_METRICS][
                  metric
              ][
                  CURRICULUM_LEARNING_DIFFICULTY_TYPE
              ]
              self.clustering_type[metric] = data_efficiency_config[
                  DATA_SAMPLING
              ][CURRICULUM_LEARNING][CURRICULUM_LEARNING_METRICS][
                  metric
              ][
                  CURRICULUM_LEARNING_CLUSTERING_TYPE
              ]
              if self.global_rank == 0:
                  if (
                      self.clustering_type[metric]
                      != CURRICULUM_LEARNING_SINGLE_CLUSTER
                  ):
                      self.curriculum_index_to_sample[
                          metric
                      ] = MMapIndexedDataset(
                          data_efficiency_config[DATA_SAMPLING][
                              CURRICULUM_LEARNING
                          ][CURRICULUM_LEARNING_METRICS][metric][
                              CURRICULUM_LEARNING_SAMPLE_PATH
                          ],
                          skip_warmup=True,
                      )
                      if (
                          self.difficulty_type[metric]
                          == CURRICULUM_LEARNING_VALUE_BASED
                      ):
                          self.curriculum_index_to_metric[
                              metric
                          ] = MMapIndexedDataset(
                              data_efficiency_config[DATA_SAMPLING][
                                  CURRICULUM_LEARNING
                              ][CURRICULUM_LEARNING_METRICS][metric][
                                  CURRICULUM_LEARNING_METRIC_PATH
                              ],
                              skip_warmup=True,
                          )
  
      # Sanity checks.
      assert self.total_samples > 0, "no sample to consume: {}".format(
          self.total_samples
      )
      assert self.micro_batch_size > 0
      assert data_parallel_size > 0
      assert self.data_parallel_rank < data_parallel_size, (
          "data_parallel_rank should be smaller than data size: {}, "
          "{}".format(self.data_parallel_rank, data_parallel_size)
      )
  
imports: import torch
  import os
  import numpy
  import deepspeed.comm
  from deepspeed.utils import logger
  from deepspeed.accelerator import get_accelerator
  from ..constants import *
  from ..curriculum_scheduler import CurriculumScheduler
  from .indexed_dataset import MMapIndexedDataset
  from .utils import create_mmap_dataset_builder, close_mmap_dataset_builder, find_fit_int_dtype
label_desc: Use image node_id 1 to create a new DeepSpeedDataSampler object from inherited base classes: object with example: obj = DeepSpeedDataSampler(data_efficiency_config, one_epoch_total_samples, micro_batch_size, data_parallel_rank, data_parallel_size, data_parallel_group, gradient_accumulation_steps, global_rank, drop_last)