code: def add(
      self,
      func,
      num_workers=None,
      batch_size=None,
      filterf=None,
      batcher=None,
      timeout=None,
  ):
      if (
          num_workers is None
          and batch_size is None
          and filterf is None
          and batcher is None
          and timeout is None
      ):
          g = self.func
          self.func = lambda x: func(g(x))
          return self
      else:
          assert self.q_out is None
          self.q_out = Queue(self.q_size)
          return BufferedPipe(
              func,
              self,
              q_in=self.q_out,
              num_workers=num_workers
              if num_workers
              else self.num_workers,
              batch_size=batch_size if batch_size else self.batch_size,
              filterf=filterf if filterf else self.filterf,
              batcher=batcher if batcher else self.batcher,
              timeout=timeout if timeout else 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 3 for calling the BufferedPipe obj's underlying member method code with example usage: obj.add(func, num_workers, batch_size, filterf, batcher, timeout) and returns: self, BufferedPipe