code: def generic_execute(function, input_gen_list, optional_params=None):
      """Given a Python `function` (taking some positional arguments) and a list of sample generators,
      execute the function twice on batches of data generated by the generator and compare to test
      the conditional execution.
  
      The function is executed both as a:
      * DALI Pipeline with conditional execution enabled. External source nodes are passed
        as positional parameters and fed with the generated batches.
      * Regular function, where we pass the batches sample-by-sample to build output batches.
  
      Parameters
      ----------
      function : callable
          function used for testing
      input_gen_list : list of sample generators
          Possibly a stateful generator
      optional_params : list of dictionaries, optional
          Optional kwargs for external source associated with given input position, by default None
      """
      if optional_params is None:
          optional_params = [{} for _ in input_gen_list]
      assert len(input_gen_list) == len(optional_params), (
          "Optional param should be provided for"
          " every external source node."
      )
      bs = 10
      iters = 5
      kwargs = {
          "batch_size": bs,
          "num_threads": 4,
          "device_id": 0,
          "prefetch_queue_depth": 1,  # so that it's easier to use external source
      }
  
      # Prepare external source nodes with placeholder names, convert
      es_inputs = [
          fn.external_source(name=f"input_{i}", **params)
          for i, params in enumerate(optional_params)
      ]
  
      pipeline_definition = pipeline_def(enable_conditionals=True)(
          function
      )
  
      def gen_batch(generator, bs, iter):
          return [
              generator(SampleInfo(bs * iter + i, i, iter, 0))
              for i in range(bs)
          ]
  
      pipe = pipeline_definition(*es_inputs, **kwargs)
      pipe.build()
  
      for iter in range(iters):
          batches = [gen_batch(gen, bs, iter) for gen in input_gen_list]
          for i, batch in enumerate(batches):
              pipe.feed_input(f"input_{i}", batch)
  
          outputs = pipe.run()
  
          baseline_outputs = []
          for inputs_i in zip(*batches):
              outputs_i = function(*inputs_i)
              # make it a tad more generic
              if not isinstance(outputs_i, tuple):
                  outputs_i = (outputs_i,)
              baseline_outputs.append(outputs_i)
  
          # Repack list of tuples into tuple of lists.
          baseline_outputs = tuple(zip(*baseline_outputs))
          # make the elements actually lists:
          baseline_outputs = (
              list(baseline) for baseline in baseline_outputs
          )
  
          for out, baseline in zip(outputs, baseline_outputs):
              check_batch(out, baseline, bs)
  
imports: from nvidia.dali.pipeline import pipeline_def, experimental
  import nvidia.dali.fn
  import nvidia.dali.types
  from nvidia.dali.types import SampleInfo
  from nvidia.dali import _conditionals
  from nvidia.dali.data_node import DataNode
  import numpy
  import os
  from test_utils import check_batch, compare_pipelines
  from nose_utils import assert_raises
  from test_utils import get_dali_extra_path
  from nose2.tools import params
  import itertools
label_desc: Use image node_id 2 for calling a global function with example usage: generic_execute(function, input_gen_list, optional_params) and returns: unknown