class_docstr: ExternalSource is a special operator that can provide data to a DALI pipeline
  from Python by several methods.
  
  The simplest and preferred way is to specify a ``source``, which can be a callable or iterable.
  
  .. note::
      :meth:`nvidia.dali.fn.external_source` operator is partially compatible with TensorFlow
      integration via :meth:`nvidia.dali.plugin.tf.experimental.DALIDatasetWithInputs`.
      Please refer to its documentation for details.
  
  .. note::
      To return a batch of copies of the same tensor, use :func:`nvidia.dali.types.Constant`,
      which is more performant.
class_docstr_tok: ["ExternalSource","is","a","special","operator","that","can","provide","data","to","a","DALI","pipeline","from","Python","by","several","methods",".","The","simplest","and","preferred","way","is","to","specify","a","``","source","``",",","which","can","be","a","callable","or","iterable",".","..","note",":",":",":","meth",":","`","nvidia.dali.fn.external_source","`","operator","is","partially","compatible","with","TensorFlow","integration","via",":","meth",":","`","nvidia.dali.plugin.tf.experimental.DALIDatasetWithInputs","`",".","Please","refer","to","its","documentation","for","details",".","..","note",":",":","To","return","a","batch","of","copies","of","the","same","tensor",",","use",":","func",":","`","nvidia.dali.types.Constant","`",",","which","is","more","performant","."]
code: def __call__(
      self,
      *,
      source=None,
      cycle=None,
      name=None,
      layout=None,
      dtype=None,
      ndim=None,
      cuda_stream=None,
      use_copy_kernel=None,
      batch=None,
      parallel=None,
      no_copy=None,
      prefetch_queue_depth=None,
      bytes_per_sample_hint=None,
      batch_info=None,
      repeat_last=False,
      **kwargs,
  ):
      """"""
      from nvidia.dali.ops import _OperatorInstance, _separate_kwargs
  
      if batch_info is None:
          batch_info = self._batch_info or False
      elif self._batch_info is not None:
          raise ValueError(
              "The argument ``batch_info`` already specified in constructor."
          )
  
      if source is None:
          if cycle is not None:
              if self._callback:
                  raise ValueError(
                      "The argument ``cycle`` can only be specified if ``source`` is an"
                      "iterable object or a generator function specified in this call. "
                      "To cycle through an iterable specified in "
                      "``__init__``, set ``cycle`` there."
                  )
              else:
                  raise ValueError(
                      "The argument ``cycle`` can only be specified if ``source`` is a "
                      "reusable iterable or a generator function."
                  )
          callback = self._callback
          source_desc = self._source_desc
      else:
          if self._callback is not None:
              raise RuntimeError(
                  "``source`` already specified in constructor."
              )
          callback, source_desc = _get_callback_from_source(
              source, cycle, self._batch_info
          )
  
          # Keep the metadata for Pipeline inspection
          self._source_desc = source_desc
  
      if callback is not None and repeat_last:
          raise ValueError(
              "``repeat_last`` must not be set when using the ``source`` argument "
              "It's usable only with manually fed ``external_source``."
          )
  
      if parallel is None:
          parallel = self._parallel or False
      elif self._parallel is not None:
          raise ValueError(
              "The argument ``parallel`` already specified in constructor."
          )
  
      if batch is None:
          batch = self._batch
      elif self._batch is not None:
          raise ValueError(
              "The argument ``batch`` already specified in constructor."
          )
  
      # By default parallel is False, so batch will be True
      if batch is None:
          batch = not parallel
  
      if prefetch_queue_depth is None:
          prefetch_queue_depth = self._prefetch_queue_depth
      elif self._prefetch_queue_depth is not None:
          raise ValueError(
              "The argument ``prefetch_queue_depth`` already specified in constructor."
          )
  
      if bytes_per_sample_hint is None:
          bytes_per_sample_hint = self._bytes_per_sample_hint
      elif self._bytes_per_sample_hint is not None:
          raise ValueError(
              "The argument ``bytes_per_sample_hint`` already specified in constructor."
          )
  
      if no_copy is None:
          no_copy = self._no_copy
      elif self._no_copy is not None:
          raise ValueError(
              "The argument ``no_copy`` already specified in constructor."
          )
  
      if parallel:
          if prefetch_queue_depth is None:
              prefetch_queue_depth = 1
          if no_copy is None:
              no_copy = True
          if not no_copy:
              raise ValueError(
                  "The argument ``no_copy`` cannot be specified to False "
                  " when used with ``parallel=True``."
              )
          if prefetch_queue_depth < 1:
              raise ValueError(
                  "``prefetch_queue_depth`` must be a positive integer, got {}.".format(
                      prefetch_queue_depth
                  )
              )
          if (
              bytes_per_sample_hint is not None
              and bytes_per_sample_hint < 1
          ):
              raise ValueError(
                  f"``bytes_per_sample_hint`` must be a positive integer, "
                  f"got {bytes_per_sample_hint}."
              )
          if source_desc.kind == _SourceKind.CALLABLE:
              if not source_desc.has_inputs:
                  raise TypeError(
                      "Callable passed to External Source in parallel mode "
                      "(when `parallel=True`) must accept exactly one argument: "
                      "`nvidia.dali.types.SampleInfo` if run with `batch=False` or "
                      "either `nvidia.dali.types.BatchInfo` or integer that "
                      "represents the index of the batch within the epoch if `batch=True`. "
                      "Got a callable that does not accept arguments instead."
                  )
          elif not batch:
              if source_desc.kind == _SourceKind.ITERABLE:
                  what = "an iterable"
              else:
                  what = "a generator function"
              raise TypeError(
                  "Parallel external source with {} must be run in a batch mode "
                  "(specify `batch=True` in the external source definition and make sure "
                  "your source returns batches)".format(what)
              )
      else:
          for kwarg_value, kwarg_name in (
              (prefetch_queue_depth, "prefetch_queue_depth"),
              (bytes_per_sample_hint, "bytes_per_sample_hint"),
          ):
              if kwarg_value is not None:
                  raise ValueError(
                      f"The argument `{kwarg_name}` is valid only for "
                      "parallel external sources (when ``parallel`` is True)."
                  )
  
      if self._layout is not None:
          if layout is not None:
              raise RuntimeError(
                  "``layout`` already specified in constructor."
              )
          else:
              layout = self._layout
  
      if self._dtype is not None:
          if dtype is not None:
              raise RuntimeError(
                  "``dtype`` already specified in constructor."
              )
          else:
              dtype = self._dtype
  
      if self._ndim is not None:
          if ndim is not None:
              raise RuntimeError(
                  "``ndim`` already specified in constructor."
              )
          else:
              ndim = self._ndim
  
      if self._cuda_stream is not None:
          if cuda_stream is not None:
              raise RuntimeError(
                  "``cuda_stream`` already specified in constructor."
              )
          else:
              cuda_stream = self._cuda_stream
  
      if self._use_copy_kernel is not None:
          if use_copy_kernel is not None:
              raise RuntimeError(
                  "``use_copy_kernel`` already specified in constructor."
              )
          else:
              use_copy_kernel = self._use_copy_kernel
  
      if name is None:
          name = self._name
      else:
          self._name = name
  
      if name is not None and self._num_outputs is not None:
          raise RuntimeError(
              "``num_outputs`` is not compatible with named ``ExternalSource``."
          )
  
      group_common_kwargs = {
          "cuda_stream": cuda_stream,
          "use_copy_kernel": use_copy_kernel,
          "batch": batch,
          "batch_info": batch_info,
          "parallel": parallel,
          "prefetch_queue_depth": prefetch_queue_depth,
          "bytes_per_sample_hint": bytes_per_sample_hint,
      }
  
      if self._num_outputs is not None:
          outputs = []
          kwargs = {"no_copy": no_copy}
          group = _ExternalSourceGroup(
              callback, source_desc, True, **group_common_kwargs
          )
          for i in range(self._num_outputs):
              if dtype is not None:
                  if isinstance(dtype, (list, tuple)):
                      kwargs["dtype"] = (
                          dtype[i]
                          if i < len(dtype)
                          else nvidia.dali.types.DALIDataType.NO_TYPE
                      )
                  else:
                      kwargs["dtype"] = dtype
              if ndim is not None:
                  if isinstance(ndim, (list, tuple)):
                      kwargs["ndim"] = (
                          ndim[i] if i < len(ndim) else None
                      )
                  else:
                      kwargs["ndim"] = ndim
              this_layout = None
              if layout is not None:
                  if isinstance(layout, (list, tuple)):
                      this_layout = layout[i] if i < len(layout) else ""
                  else:
                      this_layout = layout
                  kwargs["layout"] = this_layout
  
              args, arg_inputs = _separate_kwargs(kwargs)
              op_instance = _OperatorInstance(
                  [], arg_inputs, args, {}, self
              )
              op_instance._callback = callback
              op_instance._output_index = i
              op_instance._group = group
              op_instance._layout = this_layout
              op_instance._batch = batch
              group.append(op_instance)
              outputs.append(op_instance.unwrapped_outputs)
  
          return outputs
      else:
          if name is not None:
              kwargs["name"] = name
          if no_copy is not None:
              kwargs["no_copy"] = no_copy
          if dtype is not None:
              kwargs["dtype"] = dtype
          if ndim is not None:
              kwargs["ndim"] = ndim
          if layout is not None:
              kwargs["layout"] = layout
  
          args, arg_inputs = _separate_kwargs(kwargs)
          op_instance = _OperatorInstance(
              [], arg_inputs, args, {}, self
          )
          op_instance._callback = callback
          op_instance._output_index = None
          op_instance._group = _ExternalSourceGroup(
              callback,
              source_desc,
              False,
              [op_instance],
              **group_common_kwargs,
          )
          op_instance._layout = layout
          op_instance._batch = batch
  
          return op_instance.unwrapped_outputs
  
imports: from nvidia.dali import backend
  from nvidia.dali import tensors
  from nvidia.dali import types
  from nvidia.dali._multiproc.messages import TaskArgs, SampleRange
  import nvidia.dali.types
  from nvidia.dali._utils.external_source_impl import get_callback_from_source, accepted_arg_count, SourceKind
label_desc: Use image node_id 6 for calling the ExternalSource obj's underlying member method code with example usage: obj.__call__() and returns: outputs, op_instance