class_docstr: A common dataset object shared by most of NeMo multimodal models.
code: def __init__(
      self,
      dataset_cfg,
      map_fn: Callable,
      compose_fn: Union[Callable, List[Callable]],
      consumed_samples: int,
      filter_fn: Callable = None,
      gen_cfg=None,
      decode_fn: Callable = None,
      is_train=True,
  ):
      super().__init__()
      self.dataset_cfg = dataset_cfg
      self.num_workers = dataset_cfg.num_workers
      self.world_size = get_world_size()
      self.webdata_cfg = dataset_cfg.webdataset
      self.infinite_sampler = self.webdata_cfg.get(
          "infinite_sampler", False
      )
      self.gen_cfg = gen_cfg
      self.consumed_samples = consumed_samples
  
      self.local_root_path = self.webdata_cfg.local_root_path
      if is_train:
          dataset_path = dataset_cfg.train.dataset_path
          self.augmentations = dataset_cfg.train.get(
              "augmentations", None
          )
          self.filterings = dataset_cfg.train.get("filterings", None)
      else:
          dataset_path = dataset_cfg.validation.dataset_path
          self.augmentations = dataset_cfg.validation.get(
              "augmentations", None
          )
          self.filterings = dataset_cfg.validation.get(
              "filterings", None
          )
  
      if "boto3" in dataset_cfg:
          logging.info(
              f"Init boto3 using credentials file at {dataset_cfg.boto3.credentials_file}"
          )
          self.use_boto3 = True
          assert dataset_cfg.boto3.credentials_file is not None
          with open(dataset_cfg.boto3.credentials_file) as fin:
              self.credentials = json.load(fin)
          config = Config(
              connect_timeout=30,
              signature_version="s3",
              retries={"max_attempts": 999999},
          )
          self.s3 = boto3.client(
              "s3", **self.credentials, config=config
          )
          self.bucket = dataset_cfg.boto3.bucket
          self.local_root_path = ""
      else:
          logging.info(
              f"Read Webdataset locally. Data stores at {self.local_root_path}"
          )
          self.use_boto3 = False
          self.s3 = None
          self.bucket = None
  
      # wdinfo in a dict containing webdata information
      self.wdinfo = dict()
      if dataset_path[0].endswith(".pkl"):
          for dset_info_path in dataset_path:
              with open(dset_info_path, "rb") as fp:
                  dset_info = pickle.load(fp)
                  if "tar_files" not in self.wdinfo:
                      self.wdinfo["tar_files"] = dset_info["tar_files"]
                      self.wdinfo["total_key_count"] = dset_info[
                          "total_key_count"
                      ]
                      self.wdinfo["chunk_size"] = dset_info[
                          "chunk_size"
                      ]
                  else:
                      self.wdinfo["tar_files"].extend(
                          dset_info["tar_files"]
                      )
                      self.wdinfo["total_key_count"] += dset_info[
                          "total_key_count"
                      ]
          train_info = self.wdinfo
      else:
          train_info = self.wdinfo
          train_info["tar_files"] = map(
              wds.shardlists.expand_urls, dataset_path
          )
          train_info["tar_files"] = list(
              itertools.chain.from_iterable(train_info["tar_files"])
          )
          train_info["chunk_size"] = self.webdata_cfg.get(
              "chunk_size", 1000
          )
          train_info["total_key_count"] = train_info[
              "chunk_size"
          ] * len(train_info["tar_files"])
  
      self.data_parallel_size = (
          parallel_state.get_data_parallel_world_size()
      )
      chunk_size = train_info["chunk_size"]
  
      num_workers = dataset_cfg.get("num_workers") or 1
      self.consumed_urls = (
          consumed_samples
          // (self.data_parallel_size * num_workers)
          // chunk_size
          * (self.data_parallel_size * num_workers)
      )
      self.consumed_samples = self.consumed_urls * chunk_size
      self.skip_ahead = consumed_samples - self.consumed_samples
  
      decode_fn = pil_loader if decode_fn is None else decode_fn
      shards_train_list = train_info["tar_files"]
      num_shards = len(shards_train_list)
      assert num_shards > 0, "Did not find any training data."
  
      # Shuffle buffer:
      shuffle_buffer_size = train_info["chunk_size"]
  
      if self.filterings is not None:
          # TODO : Not a good way of estimating filtering (We expect user to give estimated portion)
          # We should estimate in someway. This is anyway used only in progress bar
          logging.info(
              f"Estimated {self.filterings.estimated_portion} will be remaining after filtering"
          )
          train_info["total_key_count"] = int(
              train_info["total_key_count"]
              * self.filterings.estimated_portion
          )
  
      # WDS Dataset Pipeline
      # DetShuffle -> Decode -> Filter -> Map -> Compose
      train_dataset, epoch = self._get_webdataset_and_epoch()
      train_dataset = train_dataset.compose(
          detshuffle2(bufsize=shuffle_buffer_size, epoch=epoch)
      )
      train_dataset = train_dataset.decode(
          decode_fn, handler=warn_and_continue
      )
  
      if self.filterings is not None:
          if self.filterings.resolution is not None:
              train_dataset = train_dataset.select(filter_fn)
  
      train_dataset = train_dataset.map(
          map_fn, handler=warn_and_continue
      )
      if not isinstance(compose_fn, list):
          compose_fn = [compose_fn]
      for fn in compose_fn:
          train_dataset = train_dataset.compose(fn)
      train_dataset.total_images = train_info["total_key_count"]
  
      if train_info["total_key_count"] != train_info[
          "chunk_size"
      ] * len(train_info["tar_files"]):
          logging.warning(
              "Total image count is not equal to chunk_size * number of tar files."
          )
  
      if self.infinite_sampler:
          (
              rank,
              world_size,
              worker_id,
              num_workers,
          ) = pytorch_worker_info()
          nbatches = (
              train_dataset.total_images
              // world_size
              // self.num_workers
          )
          logging.info(
              f"Setting nbatches={nbatches} for infinite sampler. world_size={world_size}"
          )
          train_dataset = train_dataset.with_epoch(nbatches=nbatches)
  
      logging.info("Total number of training shards: %d", num_shards)
      logging.info(
          "Total training key count: %d", train_dataset.total_images
      )
  
      self._dataset = train_dataset
  
imports: import io
  import itertools
  import json
  import os
  import pickle
  import random
  import re
  from typing import Callable, List, Union
  import boto3
  import torch.distributed
  from botocore.config import Config
  from PIL import Image
  from nemo.collections.multimodal.data.common.data_samplers import SharedEpoch, WDSUrlsRandomSampler
  from nemo.collections.multimodal.data.common.webdataset_s3 import WebDataset
  from nemo.collections.nlp.modules.common.megatron.utils import ApexGuardDefaults
  from nemo.core.classes import IterableDataset
  from nemo.utils import logging
label_desc: Use image node_id 1 to create a new WebDatasetCommon object from inherited base classes: NeMoIterableDataset with example: obj = WebDatasetCommon(dataset_cfg, map_fn, compose_fn, consumed_samples, filter_fn, gen_cfg, decode_fn, is_train)