code: def test_external_source_with_sample_info():
      batch_size = 7
      for attempt in range(10):
          pipe = Pipeline(batch_size, 3, 0)
  
          def src(si):
              assert (
                  si.idx_in_epoch
                  == batch_size * si.iteration + si.idx_in_batch
              )
              return make_array(
                  [
                      attempt * 100
                      + si.iteration * 10
                      + si.idx_in_batch
                      + 1.5
                  ],
                  dtype=datapy.float32,
              )
  
          pipe.set_outputs(fn.external_source(src, batch=False))
          pipe.build()
  
          for i in range(10):
              batch = [
                  np.array(
                      [attempt * 100 + i * 10 + s + 1.5],
                      dtype=np.float32,
                  )
                  for s in range(batch_size)
              ]
              check_output(pipe.run(), batch)
  
imports: import functools
  import numpy
  import nvidia.dali.fn
  import nvidia.dali.ops
  from nvidia.dali import Pipeline
  from nose_utils import assert_raises
  from test_utils import check_output
label_desc: Use image node_id 22 for calling a global function with example usage: test_external_source_with_sample_info() and returns: make_array