code: def get_external_source_pipe(es_args, dtype, es_device):
      def get_pipeline_desc(
          batch_size,
          num_threads,
          device,
          device_id,
          shard_id,
          num_shards,
          def_for_dataset,
      ):
          pipe = Pipeline(batch_size, num_threads, device_id)
          with pipe:
              # Our callbacks may have state, to be able to run it twice, once in Dataset and once
              # with baseline test, we need to make a copy to preserve that state.
              es = fn.external_source(
                  device=es_device, **copy.deepcopy(es_args)
              )
              if device == "gpu" and es_device == "cpu":
                  es = es.gpu()
              pad = fn.pad(es, device=device)
              pipe.set_outputs(pad)
          return pipe, None, dtype
  
      return get_pipeline_desc
  
imports: import copy
  import numpy
  import nvidia.dali.plugin.tf
  import tensorflow
  from nvidia.dali import Pipeline, fn
  from test_utils import RandomlyShapedDataIterator
label_desc: Use image node_id 7 for calling a global function with example usage: get_external_source_pipe(es_args, dtype, es_device) and returns: get_pipeline_desc, pipe,  None,  dtype