code: def _run_test(
      device,
      batch_size,
      out_dim,
      in_dim,
      in_dtype,
      out_dtype,
      M_kind,
      T_kind,
  ):
      pipe = dali.pipeline.Pipeline(
          batch_size=batch_size, num_threads=4, device_id=0, seed=1234
      )
      with pipe:
          X = fn.external_source(
              source=get_data_source(batch_size, in_dim, in_dtype),
              device=device,
              layout="NX",
          )
          M = None
          T = None
          MT = None
          if T_kind == "fused":
              MT = make_param(M_kind, [out_dim, in_dim + 1])
          else:
              M = make_param(M_kind, [out_dim, in_dim])
              T = make_param(T_kind, [out_dim])
  
          Y = fn.coord_transform(
              X,
              MT=MT.flatten().tolist()
              if isinstance(MT, np.ndarray)
              else MT,
              M=M.flatten().tolist()
              if isinstance(M, np.ndarray)
              else M,
              T=T.flatten().tolist()
              if isinstance(T, np.ndarray)
              else T,
              dtype=dali_type(out_dtype),
          )
          if M is None:
              M = 1
          if T is None:
              T = 0
          if MT is None:
              MT = 0
  
          M, T, MT = (
              x
              if isinstance(x, dali.data_node.DataNode)
              else dali.types.Constant(x, dtype=dali.types.FLOAT)
              for x in (M, T, MT)
          )
  
          pipe.set_outputs(X, Y, M, T, MT)
  
      pipe.build()
      for iter in range(3):
          outputs = pipe.run()
          outputs = [
              x.as_cpu() if hasattr(x, "as_cpu") else x for x in outputs
          ]
          ref = []
          scale = 1
          for idx in range(batch_size):
              X = outputs[0].at(idx)
              if T_kind == "fused":
                  MT = outputs[4].at(idx)
                  if MT.size == 1:
                      M = MT
                      T = 0
                  else:
                      M = MT[:, :-1]
                      T = MT[:, -1]
              else:
                  M = outputs[2].at(idx)
                  T = outputs[3].at(idx)
  
              if M.size == 1:
                  Y = X.astype(np.float32) * M + T
              else:
                  Y = np.matmul(X.astype(np.float32), M.transpose()) + T
  
              if np.issubdtype(out_dtype, np.integer):
                  info = np.iinfo(out_dtype)
                  Y = Y.clip(info.min, info.max)
  
              ref.append(Y)
              scale = (
                  max(scale, np.max(np.abs(Y)) - np.min(np.abs(Y)))
                  if Y.size > 0
                  else 1
              )
          avg = 1e-6 * scale
          eps = 1e-6 * scale
          if out_dtype != np.float32:  # headroom for rounding
              avg += 0.33
              eps += 0.5
          check_batch(
              outputs[1],
              ref,
              batch_size,
              eps,
              eps,
              expected_layout="NX",
          )
  
file_path: DALI/dali/test/python/operator_1/test_coord_transform.py
imports: import random
  import numpy
  import nvidia.dali
  import nvidia.dali.fn
  from test_utils import check_batch, dali_type
  from sequences_test_utils import ArgData, ArgDesc, sequence_suite_helper, ArgCb
label_desc: Use image node_id 5 for calling a global function with example usage: _run_test(device, batch_size, out_dim, in_dim, in_dtype, out_dtype, M_kind, T_kind) without return types