class_docstr: Helper class that constructs a tarred dataset from scratch, or concatenates tarred datasets
  together and constructs manifests for them.
docstr: Creates a new tarred dataset from a given manifest file.
  
  Args:
      manifest_path: Path to the original ASR manifest.
      target_dir: Output directory.
      num_workers: Integer denoting number of parallel worker processes which will write tarfiles.
          Defaults to 1 - which denotes sequential worker process.
  
  Output:
      Writes tarfiles, along with the tarred dataset compatible manifest file.
      Also preserves a record of the metadata used to construct this tarred dataset.
docstr_tok: ["Creates","a","new","tarred","dataset","from","a","given","manifest","file",".","Args",":","manifest_path",":","Path","to","the","original","ASR","manifest",".","target_dir",":","Output","directory",".","num_workers",":","Integer","denoting","number","of","parallel","worker","processes","which","will","write","tarfiles",".","Defaults","to","1","-","which","denotes","sequential","worker","process",".","Output",":","Writes","tarfiles",",","along","with","the","tarred","dataset","compatible","manifest","file",".","Also","preserves","a","record","of","the","metadata","used","to","construct","this","tarred","dataset","."]
code: def create_new_dataset(
      self,
      manifest_path: str,
      target_dir: str = "./tarred/",
      num_workers: int = 0,
  ):
      """
      Creates a new tarred dataset from a given manifest file.
  
      Args:
          manifest_path: Path to the original ASR manifest.
          target_dir: Output directory.
          num_workers: Integer denoting number of parallel worker processes which will write tarfiles.
              Defaults to 1 - which denotes sequential worker process.
  
      Output:
          Writes tarfiles, along with the tarred dataset compatible manifest file.
          Also preserves a record of the metadata used to construct this tarred dataset.
      """
      if self.config is None:
          raise ValueError(
              "Config has not been set. Please call `configure(config: ASRTarredDatasetConfig)`"
          )
  
      if manifest_path is None:
          raise FileNotFoundError("Manifest filepath cannot be None !")
  
      config = self.config  # type: ASRTarredDatasetConfig
  
      if not os.path.exists(target_dir):
          os.makedirs(target_dir)
  
      # Read the existing manifest
      (
          entries,
          total_duration,
          filtered_entries,
          filtered_duration,
      ) = self._read_manifest(manifest_path, config)
  
      if len(filtered_entries) > 0:
          print(
              f"Filtered {len(filtered_entries)} files which amounts to {filtered_duration} seconds of audio."
          )
      print(
          f"After filtering, manifest has {len(entries)} files which amounts to {total_duration} seconds of audio."
      )
  
      if len(entries) == 0:
          print(
              "No tarred dataset was created as there were 0 valid samples after filtering!"
          )
          return
      if config.shuffle:
          random.seed(config.shuffle_seed)
          print("Shuffling...")
          if config.keep_files_together:
              filename_entries = defaultdict(list)
              for ent in entries:
                  filename_entries[ent["audio_filepath"]].append(ent)
              filenames = list(filename_entries.keys())
              random.shuffle(filenames)
              shuffled_entries = []
              for filename in filenames:
                  shuffled_entries += filename_entries[filename]
              entries = shuffled_entries
          else:
              random.shuffle(entries)
  
      # Create shards and updated manifest entries
      print(f"Number of samples added : {len(entries)}")
      print(f"Remainder: {len(entries) % config.num_shards}")
  
      start_indices = []
      end_indices = []
      # Build indices
      for i in range(config.num_shards):
          start_idx = (len(entries) // config.num_shards) * i
          end_idx = start_idx + (len(entries) // config.num_shards)
          print(f"Shard {i} has entries {start_idx} ~ {end_idx}")
          files = set()
          for ent_id in range(start_idx, end_idx):
              files.add(entries[ent_id]["audio_filepath"])
          print(f"Shard {i} contains {len(files)} files")
          if i == config.num_shards - 1:
              # We discard in order to have the same number of entries per shard.
              print(
                  f"Have {len(entries) - end_idx} entries left over that will be discarded."
              )
  
          start_indices.append(start_idx)
          end_indices.append(end_idx)
  
      manifest_folder, _ = os.path.split(manifest_path)
  
      with Parallel(
          n_jobs=num_workers, verbose=config.num_shards
      ) as parallel:
          # Call parallel tarfile construction
          new_entries_list = parallel(
              delayed(self._create_shard)(
                  entries[start_idx:end_idx],
                  target_dir,
                  i,
                  manifest_folder,
              )
              for i, (start_idx, end_idx) in enumerate(
                  zip(start_indices, end_indices)
              )
          )
  
      if config.shard_manifests:
          sharded_manifests_dir = target_dir + "/sharded_manifests"
          if not os.path.exists(sharded_manifests_dir):
              os.makedirs(sharded_manifests_dir)
  
          for manifest in new_entries_list:
              shard_id = manifest[0]["shard_id"]
              new_manifest_shard_path = os.path.join(
                  sharded_manifests_dir, f"manifest_{shard_id}.json"
              )
              with open(
                  new_manifest_shard_path, "w", encoding="utf-8"
              ) as m2:
                  for entry in manifest:
                      json.dump(entry, m2)
                      m2.write("\n")
  
      # Flatten the list of list of entries to a list of entries
      new_entries = [
          sample for manifest in new_entries_list for sample in manifest
      ]
      del new_entries_list
  
      print("Total number of entries in manifest :", len(new_entries))
  
      # Write manifest
      new_manifest_path = os.path.join(
          target_dir, "tarred_audio_manifest.json"
      )
      with open(new_manifest_path, "w", encoding="utf-8") as m2:
          for entry in new_entries:
              json.dump(entry, m2)
              m2.write("\n")
  
      # Write metadata (default metadata for new datasets)
      new_metadata_path = os.path.join(target_dir, "metadata.yaml")
      metadata = ASRTarredDatasetMetadata()
  
      # Update metadata
      metadata.dataset_config = config
      metadata.num_samples_per_shard = (
          len(new_entries) // config.num_shards
      )
  
      # Write metadata
      metadata_yaml = OmegaConf.structured(metadata)
      OmegaConf.save(metadata_yaml, new_metadata_path, resolve=True)
  
imports: import argparse
  import copy
  import json
  import os
  import random
  import tarfile
  from collections import defaultdict
  from dataclasses import dataclass, field
  from datetime import datetime
  from io import BytesIO
  from typing import Any, List, Optional
  import numpy
  import soundfile
  from joblib import Parallel, delayed
  from omegaconf import DictConfig, OmegaConf, open_dict
label_desc: Use image node_id 3 for calling the ASRTarredDatasetBuilder obj's underlying member method code with example usage: obj.create_new_dataset(manifest_path, target_dir, num_workers) without return types