class_docstr: Helper class that constructs a tarred dataset from scratch, or concatenates tarred datasets
  together and constructs manifests for them.
docstr: Creates new tarfiles in order to create a concatenated dataset, whose manifest contains the data for
  both the original dataset as well as the new data submitted in manifest paths.
  
  Args:
      base_manifest_path: Path to the manifest file which contains the information for the original
          tarred dataset (with flattened paths).
      manifest_paths: List of one or more paths to manifest files that will be concatenated with above
          base tarred dataset.
      metadata: ASRTarredDatasetMetadata dataclass instance with overrides from command line.
      target_dir: Output directory
  
  Output:
      Writes tarfiles which with indices mapping to a "concatenated" tarred dataset,
      along with the tarred dataset compatible manifest file which includes information
      about all the datasets that comprise the concatenated dataset.
  
      Also preserves a record of the metadata used to construct this tarred dataset.
docstr_tok: ["Creates","new","tarfiles","in","order","to","create","a","concatenated","dataset",",","whose","manifest","contains","the","data","for","both","the","original","dataset","as","well","as","the","new","data","submitted","in","manifest","paths",".","Args",":","base_manifest_path",":","Path","to","the","manifest","file","which","contains","the","information","for","the","original","tarred","dataset","(","with","flattened","paths",")",".","manifest_paths",":","List","of","one","or","more","paths","to","manifest","files","that","will","be","concatenated","with","above","base","tarred","dataset",".","metadata",":","ASRTarredDatasetMetadata","dataclass","instance","with","overrides","from","command","line",".","target_dir",":","Output","directory","Output",":","Writes","tarfiles","which","with","indices","mapping","to","a","``","concatenated","''","tarred","dataset",",","along","with","the","tarred","dataset","compatible","manifest","file","which","includes","information","about","all","the","datasets","that","comprise","the","concatenated","dataset",".","Also","preserves","a","record","of","the","metadata","used","to","construct","this","tarred","dataset","."]
code: def create_concatenated_dataset(
      self,
      base_manifest_path: str,
      manifest_paths: List[str],
      metadata: ASRTarredDatasetMetadata,
      target_dir: str = "./tarred_concatenated/",
      num_workers: int = 1,
  ):
      """
      Creates new tarfiles in order to create a concatenated dataset, whose manifest contains the data for
      both the original dataset as well as the new data submitted in manifest paths.
  
      Args:
          base_manifest_path: Path to the manifest file which contains the information for the original
              tarred dataset (with flattened paths).
          manifest_paths: List of one or more paths to manifest files that will be concatenated with above
              base tarred dataset.
          metadata: ASRTarredDatasetMetadata dataclass instance with overrides from command line.
          target_dir: Output directory
  
      Output:
          Writes tarfiles which with indices mapping to a "concatenated" tarred dataset,
          along with the tarred dataset compatible manifest file which includes information
          about all the datasets that comprise the concatenated dataset.
  
          Also preserves a record of the metadata used to construct this tarred dataset.
      """
      if not os.path.exists(target_dir):
          os.makedirs(target_dir)
  
      if base_manifest_path is None:
          raise FileNotFoundError(
              "Base manifest filepath cannot be None !"
          )
  
      if manifest_paths is None or len(manifest_paths) == 0:
          raise FileNotFoundError(
              "List of additional manifest filepaths cannot be None !"
          )
  
      config = ASRTarredDatasetConfig(**(metadata.dataset_config))
  
      # Read the existing manifest (no filtering here)
      base_entries, _, _, _ = self._read_manifest(
          base_manifest_path, config
      )
      print(
          f"Read base manifest containing {len(base_entries)} samples."
      )
  
      # Precompute number of samples per shard
      if metadata.num_samples_per_shard is None:
          num_samples_per_shard = len(base_entries) // config.num_shards
      else:
          num_samples_per_shard = metadata.num_samples_per_shard
  
      print("Number of samples per shard :", num_samples_per_shard)
  
      # Compute min and max duration and update config (if no metadata passed)
      print(f"Selected max duration : {config.max_duration}")
      print(f"Selected min duration : {config.min_duration}")
  
      entries = []
      for new_manifest_idx in range(len(manifest_paths)):
          (
              new_entries,
              total_duration,
              filtered_new_entries,
              filtered_duration,
          ) = self._read_manifest(
              manifest_paths[new_manifest_idx], config
          )
  
          if len(filtered_new_entries) > 0:
              print(
                  f"Filtered {len(filtered_new_entries)} files which amounts to {filtered_duration:0.2f}"
                  f" seconds of audio from manifest {manifest_paths[new_manifest_idx]}."
              )
          print(
              f"After filtering, manifest has {len(entries)} files which amounts to {total_duration} seconds of audio."
          )
  
          entries.extend(new_entries)
  
      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...")
          random.shuffle(entries)
  
      # Drop last section of samples that cannot be added onto a chunk
      drop_count = len(entries) % num_samples_per_shard
      total_new_entries = len(entries)
      entries = entries[:-drop_count]
  
      print(
          f"Dropping {drop_count} samples from total new samples {total_new_entries} since they cannot "
          f"be added into a uniformly sized chunk."
      )
  
      # Create shards and updated manifest entries
      num_added_shards = len(entries) // num_samples_per_shard
  
      print(f"Number of samples in base dataset : {len(base_entries)}")
      print(
          f"Number of samples in additional datasets : {len(entries)}"
      )
      print(f"Number of added shards : {num_added_shards}")
      print(f"Remainder: {len(entries) % num_samples_per_shard}")
  
      start_indices = []
      end_indices = []
      shard_indices = []
      for i in range(num_added_shards):
          start_idx = (len(entries) // num_added_shards) * i
          end_idx = start_idx + (len(entries) // num_added_shards)
          shard_idx = i + config.num_shards
          print(
              f"Shard {shard_idx} has entries {start_idx + len(base_entries)} ~ {end_idx + len(base_entries)}"
          )
  
          start_indices.append(start_idx)
          end_indices.append(end_idx)
          shard_indices.append(shard_idx)
  
      manifest_folder, _ = os.path.split(base_manifest_path)
  
      with Parallel(
          n_jobs=num_workers, verbose=num_added_shards
      ) as parallel:
          # Call parallel tarfile construction
          new_entries_list = parallel(
              delayed(self._create_shard)(
                  entries[start_idx:end_idx],
                  target_dir,
                  shard_idx,
                  manifest_folder,
              )
              for i, (start_idx, end_idx, shard_idx) in enumerate(
                  zip(start_indices, end_indices, shard_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
  
      # Write manifest
      if metadata is None:
          new_version = 1  # start with `1`, where `0` indicates the base manifest + dataset
      else:
          new_version = metadata.version + 1
  
      print(
          "Total number of entries in manifest :",
          len(base_entries) + len(new_entries),
      )
  
      new_manifest_path = os.path.join(
          target_dir,
          f"tarred_audio_manifest_version_{new_version}.json",
      )
      with open(new_manifest_path, "w", encoding="utf-8") as m2:
          # First write all the entries of base manifest
          for entry in base_entries:
              json.dump(entry, m2)
              m2.write("\n")
  
          # Finally write the new entries
          for entry in new_entries:
              json.dump(entry, m2)
              m2.write("\n")
  
      # Preserve historical metadata
      base_metadata = metadata
  
      # Write metadata (updated metadata for concatenated datasets)
      new_metadata_path = os.path.join(
          target_dir, f"metadata_version_{new_version}.yaml"
      )
      metadata = ASRTarredDatasetMetadata()
  
      # Update config
      config.num_shards = config.num_shards + num_added_shards
  
      # Update metadata
      metadata.version = new_version
      metadata.dataset_config = config
      metadata.num_samples_per_shard = num_samples_per_shard
      metadata.is_concatenated_manifest = True
      metadata.created_datetime = metadata.get_current_datetime()
  
      # Attach history
      current_metadata = OmegaConf.structured(base_metadata.history)
      metadata.history = current_metadata
  
      # 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 4 for calling the ASRTarredDatasetBuilder obj's underlying member method code with example usage: obj.create_concatenated_dataset(base_manifest_path, manifest_paths, metadata, target_dir, num_workers) without return types