code: def __call__(self, data_gen):
      if self.parent:
          data_gen = self.parent(data_gen)
  
      if (
          self.num_workers == 0
          and self.batch_size == 1
          and self.q_in is None
          and self.q_out is None
      ):
          return (self._wrapped_func(x) for x in data_gen)
  
      q_in = self.q_in
      if q_in is None:
          q_in = Queue(self.q_size)
          Thread(
              target=_gen2q,
              args=(data_gen, q_in),
          ).start()
  
      q_out = self.q_out
      if q_out is None:
          q_out = Queue(self.q_size)
  
      if self.batch_size == 1:
          target = _q2q_single
          kwargs = {
              "func": self._wrapped_func,
              "queue_in": q_in,
              "queue_out": q_out,
              "filterf": self._wrapped_filter,
              "timeout": self.timeout,
          }
      else:
          target = _q2q_batch
          kwargs = {
              "func": self._wrapped_func,
              "queue_in": q_in,
              "queue_out": q_out,
              "filterf": self._wrapped_filter,
              "batch_size": self.batch_size,
              "timeout": self.timeout,
          }
  
      if self.num_workers == 0:
          self.worker_frame(target=target, kwargs=kwargs).start()
      else:
          for i in range(self.num_workers):
              self.worker_frame(target=target, kwargs=kwargs).start()
  
      if self.q_out is None:
          return _q2gen(q_out, timeout=self.timeout)
  
imports: import types
  from studio.util import logs
  from PIL import Image
  from threading import Thread
  import numpy
  import itertools
  import six
label_desc: Use image node_id 2 for calling the BufferedPipe obj's underlying member method code with example usage: obj.__call__(data_gen) and returns: unknown, _q2gen