code: def create_data_gen_pipeline(patient_data, cf, do_aug=True):
      """
      create mutli-threaded train/val/test batch generation and augmentation pipeline.
      :param patient_data: dictionary containing one dictionary per patient in the train/test subset.
      :param is_training: (optional) whether to perform data augmentation (training) or not (validation/testing)
      :return: multithreaded_generator
      """
  
      # create instance of batch generator as first element in pipeline.
      data_gen = BatchGenerator(
          patient_data, batch_size=cf.batch_size, cf=cf
      )
  
      # add transformations to pipeline.
      my_transforms = []
      if do_aug:
          mirror_transform = Mirror(axes=np.arange(2, cf.dim + 2, 1))
          my_transforms.append(mirror_transform)
          spatial_transform = SpatialTransform(
              patch_size=cf.patch_size[: cf.dim],
              patch_center_dist_from_border=cf.da_kwargs[
                  "rand_crop_dist"
              ],
              do_elastic_deform=cf.da_kwargs["do_elastic_deform"],
              alpha=cf.da_kwargs["alpha"],
              sigma=cf.da_kwargs["sigma"],
              do_rotation=cf.da_kwargs["do_rotation"],
              angle_x=cf.da_kwargs["angle_x"],
              angle_y=cf.da_kwargs["angle_y"],
              angle_z=cf.da_kwargs["angle_z"],
              do_scale=cf.da_kwargs["do_scale"],
              scale=cf.da_kwargs["scale"],
              random_crop=cf.da_kwargs["random_crop"],
          )
  
          my_transforms.append(spatial_transform)
      else:
          my_transforms.append(
              CenterCropTransform(crop_size=cf.patch_size[: cf.dim])
          )
  
      my_transforms.append(
          ConvertSegToBoundingBoxCoordinates(
              cf.dim,
              get_rois_from_seg_flag=False,
              class_specific_seg_flag=cf.class_specific_seg_flag,
          )
      )
      all_transforms = Compose(my_transforms)
      # multithreaded_generator = SingleThreadedAugmenter(data_gen, all_transforms)
      multithreaded_generator = MultiThreadedAugmenter(
          data_gen,
          all_transforms,
          num_processes=cf.n_workers,
          seeds=range(cf.n_workers),
      )
      return multithreaded_generator
  
imports: import numpy
  import os
  from collections import OrderedDict
  import pandas
  import pickle
  import time
  import subprocess
  import utils.dataloader_utils
  from batchgenerators.dataloading.data_loader import SlimDataLoaderBase
  from batchgenerators.transforms.spatial_transforms import MirrorTransform
  from batchgenerators.transforms.abstract_transforms import Compose
  from batchgenerators.dataloading.multi_threaded_augmenter import MultiThreadedAugmenter
  from batchgenerators.dataloading import SingleThreadedAugmenter
  from batchgenerators.transforms.spatial_transforms import SpatialTransform
  from batchgenerators.transforms.crop_and_pad_transforms import CenterCropTransform
  from batchgenerators.transforms.utility_transforms import ConvertSegToBoundingBoxCoordinates
label_desc: Use image node_id 4 for calling a global function with example usage: create_data_gen_pipeline(patient_data, cf, do_aug) and returns: multithreaded_generator