class_docstr: A ConcurrentVentilator handles ventilation of a pre-determined list of items to a worker pool and performs
  the ventilation concurrently in a separate thread. It will keep track of how many items are currently in the
  ventilation queue and prevent it from monotonically increasing in order to prevent boundless memory requirements.
  It allows for multiple (or infinite) iterations of ventilating the items, optionally randomizing the order of
  items being ventilated at the start of each iteration.
class_docstr_tok: ["A","ConcurrentVentilator","handles","ventilation","of","a","pre-determined","list","of","items","to","a","worker","pool","and","performs","the","ventilation","concurrently","in","a","separate","thread",".","It","will","keep","track","of","how","many","items","are","currently","in","the","ventilation","queue","and","prevent","it","from","monotonically","increasing","in","order","to","prevent","boundless","memory","requirements",".","It","allows","for","multiple","(","or","infinite",")","iterations","of","ventilating","the","items",",","optionally","randomizing","the","order","of","items","being","ventilated","at","the","start","of","each","iteration","."]
docstr: Constructor for a concurrent ventilator.
  
  :param ventilate_fn: The function to be called when ventilating. Usually the worker pool ventilate function.
  :param items_to_ventilate: (``list[dict]``) The list of items to ventilate. Each item is a ``dict`` denoting
          the ``**kwargs`` eventually passed to a worker process function
  :param iterations: (int) How many iterations through items_to_ventilate should be done and ventilated to the
          worker pool. For example if set to 2 each item in items_to_ventilate will be ventilated 2 times. If
          ``None`` is passed, the ventilator will continue ventilating forever.
  :param randomize_item_order: (``bool``) Whether to randomize the item order in items_to_ventilate. This will be
          done on every individual iteration.
  :param random_seed: (``int``) If not None: the random seed used for randomize_item_order. Default: None.
  :param max_ventilation_queue_size: (``int``) The maximum number of items to be stored in the ventilation queue.
          The higher this number, the higher potential memory requirements. By default it will use the size
          of items_to_ventilate since that can definitely be held in memory.
  :param ventilation_interval: (``float`` in seconds) How much time passes between checks on whether something
          can be ventilated (when the ventilation queue is considered full).
docstr_tok: ["Constructor","for","a","concurrent","ventilator",".",":","param","ventilate_fn",":","The","function","to","be","called","when","ventilating",".","Usually","the","worker","pool","ventilate","function",".",":","param","items_to_ventilate",":","(","``","list","[","dict","]","``",")","The","list","of","items","to","ventilate",".","Each","item","is","a","``","dict","``","denoting","the","``","*","*","kwargs","``","eventually","passed","to","a","worker","process","function",":","param","iterations",":","(","int",")","How","many","iterations","through","items_to_ventilate","should","be","done","and","ventilated","to","the","worker","pool",".","For","example","if","set","to","2","each","item","in","items_to_ventilate","will","be","ventilated","2","times",".","If","``","None","``","is","passed",",","the","ventilator","will","continue","ventilating","forever",".",":","param","randomize_item_order",":","(","``","bool","``",")","Whether","to","randomize","the","item","order","in","items_to_ventilate",".","This","will","be","done","on","every","individual","iteration",".",":","param","random_seed",":","(","``","int","``",")","If","not","None",":","the","random","seed","used","for","randomize_item_order",".","Default",":","None",".",":","param","max_ventilation_queue_size",":","(","``","int","``",")","The","maximum","number","of","items","to","be","stored","in","the","ventilation","queue",".","The","higher","this","number",",","the","higher","potential","memory","requirements",".","By","default","it","will","use","the","size","of","items_to_ventilate","since","that","can","definitely","be","held","in","memory",".",":","param","ventilation_interval",":","(","``","float","``","in","seconds",")","How","much","time","passes","between","checks","on","whether","something","can","be","ventilated","(","when","the","ventilation","queue","is","considered","full",")","."]
code: def __init__(
      self,
      ventilate_fn,
      items_to_ventilate,
      iterations=1,
      randomize_item_order=False,
      random_seed=None,
      max_ventilation_queue_size=None,
      ventilation_interval=_VENTILATION_INTERVAL,
  ):
      """
      Constructor for a concurrent ventilator.
  
      :param ventilate_fn: The function to be called when ventilating. Usually the worker pool ventilate function.
      :param items_to_ventilate: (``list[dict]``) The list of items to ventilate. Each item is a ``dict`` denoting
              the ``**kwargs`` eventually passed to a worker process function
      :param iterations: (int) How many iterations through items_to_ventilate should be done and ventilated to the
              worker pool. For example if set to 2 each item in items_to_ventilate will be ventilated 2 times. If
              ``None`` is passed, the ventilator will continue ventilating forever.
      :param randomize_item_order: (``bool``) Whether to randomize the item order in items_to_ventilate. This will be
              done on every individual iteration.
      :param random_seed: (``int``) If not None: the random seed used for randomize_item_order. Default: None.
      :param max_ventilation_queue_size: (``int``) The maximum number of items to be stored in the ventilation queue.
              The higher this number, the higher potential memory requirements. By default it will use the size
              of items_to_ventilate since that can definitely be held in memory.
      :param ventilation_interval: (``float`` in seconds) How much time passes between checks on whether something
              can be ventilated (when the ventilation queue is considered full).
      """
      super(ConcurrentVentilator, self).__init__(ventilate_fn)
  
      if iterations is not None and (
          not isinstance(iterations, int) or iterations < 1
      ):
          raise ValueError(
              "iterations must be positive integer or None"
          )
  
      if not isinstance(items_to_ventilate, list) or any(
          not isinstance(item, dict) for item in items_to_ventilate
      ):
          raise ValueError("items_to_ventilate must be a list of dicts")
  
      self._items_to_ventilate = items_to_ventilate
      self._iterations_remaining = iterations
      self._randomize_item_order = randomize_item_order
      self._random_state = np.random.RandomState(seed=random_seed)
      self._iterations = iterations
  
      # For the default max ventilation queue size we will use the size of the items to ventilate
      self._max_ventilation_queue_size = (
          max_ventilation_queue_size or len(items_to_ventilate)
      )
      self._ventilation_interval = ventilation_interval
  
      self._current_item_to_ventilate = 0
      self._ventilation_thread = None
      self._ventilated_items_count = 0
      self._processed_items_count = 0
      self._stop_requested = False
  
imports: import numpy
  import threading
  from abc import ABCMeta, abstractmethod
  from time import sleep
  import six
label_desc: Use image node_id 1 to create a new ConcurrentVentilator object from inherited base classes: Ventilator with example: obj = ConcurrentVentilator(ventilate_fn, items_to_ventilate, iterations, randomize_item_order, random_seed, max_ventilation_queue_size, ventilation_interval)