code: def __call__(self, scheduled: ScheduledTask):
      task = scheduled.task
      if task.is_sample_mode():
          data_batch = [
              self.callback(sample_info)
              for sample_info in task.sample_range
          ]
      else:
          data_batch = self.callback(*task.batch_args)
      return data_batch
  
imports: from typing import List, Optional, Tuple
  import threading
  import traceback
  import socket
  from collections import deque
  from multiprocessing import reduction
  from nvidia.dali._utils.external_source_impl import SourceKind, _is_generator_function
  from nvidia.dali._multiproc.shared_batch import SharedBatchWriter, SharedBatchMeta, BufShmChunk, assert_valid_data_type, read_shm_message, write_shm_message
  from nvidia.dali._multiproc.messages import CompletedTask, WorkerArgs, ShmMessageDesc, ScheduledTask
  from nvidia.dali._multiproc.shared_queue import Dispatcher
label_desc: Use image node_id 2 for calling the CallableSource obj's underlying member method code with example usage: obj.__call__(scheduled) and returns: data_batch