code: def test_named_tensor_arguments(op):
      ops2params = {
          fn.permute_batch: _tensor_arg_permute_batch_params,
          fn.roi_random_crop: _tensor_arg_roi_random_crop_params,
          fn.transforms.crop: _tensor_arg_transform_per_dim_params(
              "from_start"
          ),
          fn.transforms.scale: _tensor_arg_transform_per_dim_params(
              "scale"
          ),
          fn.transforms.shear: _tensor_arg_transform_per_dim_params(
              "angles"
          ),
          fn.transforms.translation: _tensor_arg_transform_per_dim_params(
              "offset"
          ),
          fn.transforms.rotation: _tensor_arg_rotate_params,
          fn.random.uniform: _tensor_arg_shape_kwarg,
          fn.random.normal: _tensor_arg_shape_kwarg,
          fn.random.coin_flip: _tensor_arg_shape_kwarg,
      }
  
      def dummy_source(batches):
          def cb():
              for batch in batches:
                  yield batch
  
          return cb
  
      def get_pipeline(
          op,
          args_batches,
          mask_batches,
          kwargs_batches,
          num_threads=4,
          device_id=0,
      ):
          max_batch_size = max(len(batch) for batch in mask_batches)
  
          @pipeline_def(
              batch_size=max_batch_size,
              num_threads=num_threads,
              device_id=device_id,
          )
          def split_pipeline():
              args = [
                  fn.external_source(dummy_source(arg_batches))
                  for arg_batches in args_batches
              ]
              mask = fn.external_source(dummy_source(mask_batches))
              kwargs = {
                  kwarg_name: fn.external_source(dummy_source(batches))
                  for kwarg_name, batches in kwargs_batches.items()
              }
              kwargs_split = {
                  kwarg_name: fn._conditional.split(
                      batch, predicate=mask
                  )
                  for kwarg_name, batch in kwargs.items()
              }
              split_args = [
                  fn._conditional.split(arg, predicate=mask)
                  for arg in args
              ]
              left_args = [left_arg for left_arg, _ in split_args]
              right_args = [right_arg for _, right_arg in split_args]
              left = op(
                  *left_args,
                  **{
                      kwarg_name: left_kwarg
                      for kwarg_name, (
                          left_kwarg,
                          _,
                      ) in kwargs_split.items()
                  },
              )
              right = op(
                  *right_args,
                  **{
                      kwarg_name: right_kwarg
                      for kwarg_name, (
                          _,
                          right_kwarg,
                      ) in kwargs_split.items()
                  },
              )
              batch = fn._conditional.merge(left, right, predicate=mask)
              return batch
  
          return split_pipeline()
  
      args_batches, mask_batches, kwargs_batches = ops2params[op]()
      pipe = get_pipeline(
          op=op,
          args_batches=args_batches,
          mask_batches=mask_batches,
          kwargs_batches=kwargs_batches,
      )
      pipe.build()
      for _ in range(len(mask_batches)):
          pipe.run()
  
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 38 for calling a global function with example usage: test_named_tensor_arguments(op) and returns: cb, split_pipeline, batch