class_docstr: ImgTextTarDataset class for datasets that store Image-Text pairs as tar files, each tar file with multiple pairs.
  
  The dataset should be stored in following format where `img_text_tar_dataset` is the location of directory that
  has all tar files.
  
  img_text_tar_dataset
  |--- 00000000_0_1000.tar.gz
  |-------- 00000000_0_image
  |-------- 00000000_0_text
  |-------- 00000000_1_image
  |-------- 00000000_1_text
  |-------- ...
  
  |--- 00000000_1000_2000.tar.gz
  |-------- 00000000_1000_image
  |-------- 00000000_1000_text
  |-------- 00000000_1001_image
  |-------- 00000000_1001_text
  |-------- ...
  
  Args:
      opts: An argparse.Namespace instance.
class_docstr_tok: ["ImgTextTarDataset","class","for","datasets","that","store","Image-Text","pairs","as","tar","files",",","each","tar","file","with","multiple","pairs",".","The","dataset","should","be","stored","in","following","format","where","`","img_text_tar_dataset","`","is","the","location","of","directory","that","has","all","tar","files",".","img_text_tar_dataset","|","--","-","00000000_0_1000.tar.gz","|","--","--","--","--","00000000_0_image","|","--","--","--","--","00000000_0_text","|","--","--","--","--","00000000_1_image","|","--","--","--","--","00000000_1_text","|","--","--","--","--","...","|","--","-","00000000_1000_2000.tar.gz","|","--","--","--","--","00000000_1000_image","|","--","--","--","--","00000000_1000_text","|","--","--","--","--","00000000_1001_image","|","--","--","--","--","00000000_1001_text","|","--","--","--","--","...","Args",":","opts",":","An","argparse.Namespace","instance","."]
docstr: Download the dataset
docstr_tok: ["Download","the","dataset"]
code: def _download_dataset(self) -> None:
      """Download the dataset"""
      os.makedirs(self.cache_loc, exist_ok=True)
  
      # The total number of GPUs that a task is using is equal to the world size
      world_size = getattr(self.opts, "ddp.world_size")
      if world_size is None or world_size == -1:
          if self.is_start_rank_node:
              logger.error(
                  "DDP world size should be greater than 1. Got: {}"
              )
  
      # find the number of GPUs in each node
      n_gpus_per_node = torch.cuda.device_count()
  
      # Total number of GPUs = Total number of nodes * number of GPUs per Node
      n_nodes = max(1, world_size // n_gpus_per_node)
  
      # Find the node id based on current node rank
      # node_id = current_node_rank / n_gpus_per_node
      curr_node_rank = getattr(self.opts, "ddp.rank", None)
      if curr_node_rank is None:
          if self.is_start_rank_node:
              logger.error("Node rank can't be None.")
      node_id = curr_node_rank // n_gpus_per_node
  
      # Downloading the entire dataset on each node is not feasible. Instead, for each
      # node, we will download a subset of the dataset and learn from it.
  
      # Split the dataset almost equally among all nodes. The length of this split
      # is going to be the same as the number of nodes.
      node_wise_dataset_split = np.array_split(
          self.dataset_keys, n_nodes
      )
  
      # download files corresponding to ith node
      files_node_i = node_wise_dataset_split[node_id]
  
      # Dataset is organized as a dict where key corresponds to start_index of image-text pair and
      # value corresponds to the file name.
  
      # find the start and end image-text pair indexes for node_i.
      # Note that we overlap node_i and node_i+1 by at most __OVERLAP_RATIO files
      start_idx_node_i = max(
          0,
          self.dataset_keys.index(files_node_i[0])
          - self.__OVERLAP_RATIO,
      )
      end_idx_node_i = min(
          len(self.dataset_keys),
          self.dataset_keys.index(files_node_i[-1])
          + self.__OVERLAP_RATIO,
      )
  
      # Now, download the files concurrently using each rank on node i
      # Now, download the files concurrently using each rank on node i
      indexes_to_download_node_i = self.dataset_keys[
          start_idx_node_i:end_idx_node_i
      ]
      indexes_to_download_node_i_rank_j = np.array_split(
          indexes_to_download_node_i, n_gpus_per_node
      )
      total_files_to_download = len(indexes_to_download_node_i)
      if self.is_start_rank_node:
          logger.log(
              f"Starting to downloading {total_files_to_download} files"
          )
  
      current_device = torch.cuda.current_device()
      if getattr(
          self.opts,
          "dataset.multi_modal_img_text.img_text_tar.parallel_download",
          False,
      ):
          # download concurrently using many workers for each rank
          n_cpus = resources.cpu_count()
          n_process_per_gpu = max(
              1, n_cpus // torch.cuda.device_count()
          )  # max(1, min(4, n_cpus // torch.cuda.device_count()))
          with Pool(processes=n_process_per_gpu) as pool:
              pool.starmap(
                  async_download_file_from_s3,
                  [
                      (
                          self.opts,
                          os.path.join(
                              self.s3_bucket_path,
                              self.dataset[img_text_idx],
                          ),
                          self.cache_loc,
                      )
                      for img_text_idx in indexes_to_download_node_i_rank_j[
                          current_device
                      ]
                  ],
              )
      else:
          # download sequentially (1 worker per rank)
          for count, img_text_idx in enumerate(
              indexes_to_download_node_i_rank_j[current_device]
          ):
              # Recall that dataset is organized as a dict where key corresponds to start_index of image-text pair
              # value corresponds to the tar file name.
  
              async_download_file_from_s3(
                  opts=self.opts,
                  tar_file_name=os.path.join(
                      self.s3_bucket_path, self.dataset[img_text_idx]
                  ),
                  cache_loc=self.cache_loc,
              )
  
              if count % 100 == 0 and self.is_start_rank_node:
                  n_files_downloaded = len(
                      glob.glob(f"{self.cache_loc}/*")
                  )
                  print(
                      f"Progress: {n_files_downloaded}/{total_files_to_download}",
                      end="\r",
                  )
  
      # synchronize between all DDP jobs
      if getattr(self.opts, "ddp.use_distributed", False):
          dist_barrier()
  
      if self.is_start_rank_node:
          n_files_downloaded = len(glob.glob(f"{self.cache_loc}/*"))
          logger.log(
              f"Download complete ({n_files_downloaded}/{total_files_to_download}). "
              f"Files are stored at: {self.cache_loc}"
          )
  
imports: import argparse
  import glob
  import io
  import os
  import pickle
  import tarfile
  from multiprocessing.pool import Pool
  from typing import AnyStr, Dict, Tuple
  import numpy
  import torch
  from PIL import Image, ImageFile
  from data.datasets import DATASET_REGISTRY
  from data.datasets.multi_modal_img_text.base_multi_modal_img_text import BaseMultiModalImgText
  from utils import logger, resources
  from utils.ddp_utils import dist_barrier
  from utils.download_utils import get_local_path
label_desc: Use image node_id 3 for calling the ImgTextTarDataset obj's underlying member method code with example usage: obj._download_dataset() without return types