  
  
  def total_cpus():
      """ Return total number of cpus """
      return mp.cpu_count()
  
  
  class PoolProcess():
      """ Pool multiple processes """
      def __init__(self, method, in_queue, out_queue, *args, processes=None, **kwargs):
          self._name = method.__qualname__
          logger.debug("Initializing %s: (target: '%s', processes: %s)",
                       self.__class__.__name__, self._name, processes)
  
          self.procs = self.set_procs(processes)
          ctx = mp.get_context("spawn")
          self.pool = ctx.Pool(processes=self.procs,
                               initializer=set_root_logger,
                               initargs=(logger.getEffectiveLevel(), LOG_QUEUE))
          self._method = method
          self._kwargs = self.build_target_kwargs(in_queue, out_queue, kwargs)
          self._args = args
  
          logger.debug("Initialized %s: '%s'", self.__class__.__name__, self._name)
  
      @staticmethod
      def build_target_kwargs(in_queue, out_queue, kwargs):
          """ Add standard kwargs to passed in kwargs list """
          kwargs["in_queue"] = in_queue
          kwargs["out_queue"] = out_queue
          return kwargs
  
      def set_procs(self, processes):
          """ Set the number of processes to use """
          processes = mp.cpu_count() if processes is None else processes
          running_processes = len(mp.active_children())
          avail_processes = max(mp.cpu_count() - running_processes, 1)
          processes = min(avail_processes, processes)
          logger.verbose("Processing '%s' in %s processes", self._name, processes)
          return processes
  
      def start(self):
          """ Run the processing pool """
          logging.debug("Pooling Processes: (target: '%s', args: %s, kwargs: %s)",
                        self._name, self._args, self._kwargs)
          for idx in range(self.procs):
              logger.debug("Adding process %s of %s to mp.Pool '%s'",
                           idx + 1, self.procs, self._name)
              self.pool.apply_async(self._method, args=self._args, kwds=self._kwargs)
              _launched_processes.add(self.pool)
          logging.debug("Pooled Processes: '%s'", self._name)
  
      def join(self):
          """ Join the process """
          logger.debug("Joining Pooled Process: '%s'", self._name)
          self.pool.close()
          self.pool.join()
          _launched_processes.remove(self.pool)
          logger.debug("Joined Pooled Process: '%s'", self._name)
  
  
  class SpawnProcess(mp.context.SpawnProcess):
      """ Process in spawnable context
          Must be spawnable to share CUDA across processes """
      def __init__(self, target, in_queue, out_queue, *args, **kwargs):
          name = target.__qualname__
          logger.debug("Initializing %s: (target: '%s', args: %s, kwargs: %s)",
                       self.__class__.__name__, name, args, kwargs)
          ctx = mp.get_context("spawn")
          self.event = ctx.Event()
          self.error = ctx.Event()
          kwargs = self.build_target_kwargs(in_queue, out_queue, kwargs)
          super().__init__(target=target, name=name, args=args, kwargs=kwargs)
          self.daemon = True
          logger.debug("Initialized %s: '%s'", self.__class__.__name__, name)
  
      def build_target_kwargs(self, in_queue, out_queue, kwargs):
          """ Add standard kwargs to passed in kwargs list """
          kwargs["event"] = self.event
          kwargs["error"] = self.error
          kwargs["log_init"] = set_root_logger
          kwargs["log_queue"] = LOG_QUEUE
          kwargs["log_level"] = logger.getEffectiveLevel()
          kwargs["in_queue"] = in_queue
          kwargs["out_queue"] = out_queue
          return kwargs
  
      def run(self):
          """ Add logger to spawned process """
          logger_init = self._kwargs["log_init"]
          log_queue = self._kwargs["log_queue"]
          log_level = self._kwargs["log_level"]
          logger_init(log_level, log_queue)
          super().run()
  
      def start(self):
          """ Add logging to start function """
          logger.debug("Spawning Process: (name: '%s', args: %s, kwargs: %s, daemon: %s)",
                       self._name, self._args, self._kwargs, self.daemon)
          super().start()
          _launched_processes.add(self)
          logger.debug("Spawned Process: (name: '%s', PID: %s)", self._name, self.pid)
  
      def join(self, timeout=None):
          """ Add logging to join function """
          logger.debug("Joining Process: (name: '%s', PID: %s)", self._name, self.pid)
          super().join(timeout=timeout)
          if self in _launched_processes:
              _launched_processes.remove(self)
          logger.debug("Joined Process: (name: '%s', PID: %s)", self._name, self.pid)
  
  
  class FSThread(threading.Thread):
      """ Subclass of thread that passes errors back to parent """
      def __init__(self, group=None, target=None, name=None,  # pylint: disable=too-many-arguments
 
</s> remove     # <<< OVERRIDE METHODS >>> #
    # These methods must be overriden when creating a plugin
    def initialize(self, *args, **kwargs):
        """ Inititalize the aligner
            Tasks to be run before any alignments are performed.
            Override for specific detector """
        logger.debug("_base initialize %s: (PID: %s, args: %s, kwargs: %s)",
                     self.__class__.__name__, os.getpid(), args, kwargs)
        self.init = kwargs["event"]
        self.error = kwargs["error"]
        self.queues["in"] = kwargs["in_queue"]
        self.queues["out"] = kwargs["out_queue"]

    def align_image(self, detected_face, image):
        """ Align the incoming image for feeding into aligner
            Override for aligner specific processing """
        raise NotImplementedError

    def predict_landmarks(self, feed_dict):
        """ Predict the 68 point landmarks
            Override for aligner specific landmark prediction """
        raise NotImplementedError

    # <<< GET MODEL >>> #
    @staticmethod
    def get_model(git_model_id, model_filename):
        """ Check if model is available, if not, download and unzip it """
        if model_filename is None:
            logger.debug("No model_filename specified. Returning None")
            return None
        if git_model_id is None:
            logger.debug("No git_model_id specified. Returning None")
            return None
        cache_path = os.path.join(os.path.dirname(__file__), ".cache")
        model = GetModel(model_filename, cache_path, git_model_id)
        return model.model_path

    # <<< ALIGNMENT WRAPPER >>> #
    def run(self, *args, **kwargs):
        """ Parent align process.
            This should always be called as the entry point so exceptions
            are passed back to parent.
            Do not override """
        try:
            self.align(*args, **kwargs)
        except Exception:  # pylint:disable=broad-except
            logger.error("Caught exception in child process: %s", os.getpid())
            # Display traceback if in initialization stage
            if not self.init.is_set():
                logger.exception("Traceback:")
            tb_buffer = StringIO()
            traceback.print_exc(file=tb_buffer)
            exception = {"exception": (os.getpid(), tb_buffer)}
            self.queues["out"].put(exception)
            exit(1)

    def align(self, *args, **kwargs):
        """ Process landmarks """
        if not self.init:
            self.initialize(*args, **kwargs)
        logger.debug("Launching Align: (args: %s kwargs: %s)", args, kwargs)

        for item in self.get_item():
 </s> add     # << QUEUE METHODS >>> #
    def get_batch(self, queue):
        """ Get items for inputting into the aligner from the queue in batches

        Items are returned from the ``queue`` in batches of
        :attr:`~plugins.extract._base.Extractor.batchsize`

        To ensure consistent batchsizes for aligner the items are split into separate items for
        each :class:`lib.faces_detect.DetectedFace` object.

        Remember to put ``'EOF'`` to the out queue after processing
        the final batch

        Outputs items in the following format. All lists are of length
        :attr:`~plugins.extract._base.Extractor.batchsize`:

        >>> {'filename': [<filenames of source frames>],
        >>>  'image': [<source images>],
        >>>  'detected_faces': [[<lib.faces_detect.DetectedFace objects]]}

        Parameters
        ----------
        queue : queue.Queue()
            The ``queue`` that the plugin will be fed from.

        Returns
        -------
        exhausted, bool
            ``True`` if queue is exhausted, ``False`` if not
        batch, dict
            A dictionary of lists of :attr:`~plugins.extract._base.Extractor.batchsize`:
        """
        exhausted = False
        batch = dict()
        idx = 0
        while idx < self.batchsize:
            item = self._collect_item(queue) </s> remove class Detector():
    """ Detector object """
    def __init__(self, loglevel, configfile=None,  # pylint:disable=too-many-arguments
                 git_model_id=None, model_filename=None, rotation=None, min_size=0):
        logger.debug("Initializing %s: (loglevel: %s, configfile: %s, git_model_id: %s, "
                     "model_filename: %s, rotation: %s, min_size: %s)",
                     self.__class__.__name__, loglevel, configfile, git_model_id,
                     model_filename, rotation, min_size)
        self.config = get_config(".".join(self.__module__.split(".")[-2:]), configfile=configfile)
        self.loglevel = loglevel
        self.rotation = self.get_rotation_angles(rotation)
        self.min_size = min_size
        self.parent_is_pool = False
        self.init = None
        self.error = None

        # The input and output queues for the plugin.
        # See lib.queue_manager.QueueManager for getting queues
        self.queues = {"in": None, "out": None}

        #  Path to model if required
        self.model_path = self.get_model(git_model_id, model_filename)

        # Target image size for passing images through the detector
        # Set to tuple of dimensions (x, y) or int of pixel count
        self.target = None

        # Approximate VRAM used for the set target. Used to calculate
        # how many parallel processes / batches can be run.
        # Be conservative to avoid OOM.
        self.vram = None

        # Set to true if the plugin supports PlaidML
        self.supports_plaidml = False

        # For detectors that support batching, this should be set to
        # the calculated batch size that the amount of available VRAM
        # will support. It is also used for holding the number of threads/
        # processes for parallel processing plugins
        self.batch_size = 1
 </s> add  </s> remove     # <<< OVERRIDE METHODS >>> #
    def initialize(self, *args, **kwargs):
        """ Inititalize the detector
            Tasks to be run before any detection is performed.
            Override for specific detector """
        logger.debug("initialize %s (PID: %s, args: %s, kwargs: %s)",
                     self.__class__.__name__, os.getpid(), args, kwargs)
        self.init = kwargs.get("event", False)
        self.error = kwargs.get("error", False)
        self.queues["in"] = kwargs["in_queue"]
        self.queues["out"] = kwargs["out_queue"]

    def detect_faces(self, *args, **kwargs):
        """ Detect faces in rgb image
            Override for specific detector
            Must return a list of bounding box dicts (See module docstring)"""
        try:
            if not self.init:
                self.initialize(*args, **kwargs)
        except ValueError as err:
            logger.error(err)
            exit(1)
        logger.debug("Detecting Faces (args: %s, kwargs: %s)", args, kwargs)

    # <<< GET MODEL >>> #
    @staticmethod
    def get_model(git_model_id, model_filename):
        """ Check if model is available, if not, download and unzip it """
        if model_filename is None:
            logger.debug("No model_filename specified. Returning None")
            return None
        if git_model_id is None:
            logger.debug("No git_model_id specified. Returning None")
            return None
        cache_path = os.path.join(os.path.dirname(__file__), ".cache")
        model = GetModel(model_filename, cache_path, git_model_id)
        return model.model_path

    # <<< DETECTION WRAPPER >>> #
    def run(self, *args, **kwargs):
        """ Parent detect process.
            This should always be called as the entry point so exceptions
            are passed back to parent.
            Do not override """
        try:
            logger.debug("Executing detector run function")
            self.detect_faces(*args, **kwargs)
        except Exception as err:  # pylint: disable=broad-except
            logger.error("Caught exception in child process: %s: %s", os.getpid(), str(err))
            # Display traceback if in initialization stage
            if not self.init.is_set():
                logger.exception("Traceback:")
            tb_buffer = StringIO()
            traceback.print_exc(file=tb_buffer)
            logger.trace(tb_buffer.getvalue())
            exception = {"exception": (os.getpid(), tb_buffer)}
            self.queues["out"].put(exception)
            exit(1)
 </s> add     # <<< QUEUE METHODS >>> #
    def get_batch(self, queue):
        """ Get items for inputting to the detector plugin in batches

        Items are returned from the ``queue`` in batches of
        :attr:`~plugins.extract._base.Extractor.batchsize`

        Remember to put ``'EOF'`` to the out queue after processing
        the final batch

        Outputs items in the following format. All lists are of length
        :attr:`~plugins.extract._base.Extractor.batchsize`:

        >>> {'filename': [<filenames of source frames>],
        >>>  'image': [<source images>],
        >>>  'scaled_image': <np.array of images standardized for prediction>,
        >>>  'scale': [<scaling factors for each image>],
        >>>  'pad': [<padding for each image>],
        >>>  'detected_faces': [[<lib.faces_detect.DetectedFace objects]]}

        Parameters
        ----------
        queue : queue.Queue()
            The ``queue`` that the batch will be fed from. This will be a queue that loads
            images.

        Returns
        -------
        exhausted, bool
            ``True`` if queue is exhausted, ``False`` if not.
        batch, dict
            A dictionary of lists of :attr:`~plugins.extract._base.Extractor.batchsize`.
        """
        exhausted = False
        batch = dict()
        for _ in range(self.batchsize):
            item = self._get_item(queue)
            if item == "EOF":
                exhausted = True
                break
            for key, val in item.items():
                batch.setdefault(key, []).append(val)
            scaled_image, scale, pad = self._compile_detection_image(item["image"])
            batch.setdefault("scaled_image", []).append(scaled_image)
            batch.setdefault("scale", []).append(scale)
            batch.setdefault("pad", []).append(pad)
        if batch:
            batch["scaled_image"] = np.array(batch["scaled_image"], dtype="float32")
            logger.trace("Returning batch: %s", {k: v.shape if isinstance(v, np.ndarray) else v
                                                 for k, v in batch.items() if k != "image"})
        else:
            logger.trace(item)
        return exhausted, batch </s> remove def get_config(plugin_name, configfile=None):
    """ Return the config for the requested model """
    return Config(plugin_name, configfile=configfile).config_dict
 </s> add     def __init__(self, git_model_id=None, model_filename=None,
                 configfile=None, rotation=None, min_size=0):
        logger.debug("Initializing %s: (rotation: %s, min_size: %s)", self.__class__.__name__,
                     rotation, min_size)
        super().__init__(git_model_id,
                         model_filename,
                         configfile=configfile)
        self.rotation = self._get_rotation_angles(rotation)
        self.min_size = min_size </s> remove             if not self.model.is_gpu:
                alloc = 2048
                logger.warning("Using CPU")
            else:
                logger.debug("Using GPU")
                alloc = vram_free
            logger.debug("Allocated for Tensorflow: %sMB", alloc)

            if self.min_vram < alloc < self.vram:
                self.batch_size = 1
                logger.warning("You are running s3fd with %sMB VRAM. The model is optimized for "
                               "%sMB VRAM. Detection should still run but you may get "
                               "warnings/errors", int(alloc), self.vram)
            else:
                self.batch_size = int(alloc / self.vram)
            if self.batch_size < 1:
                raise ValueError("Insufficient VRAM available to continue "
                                 "({}MB)".format(int(alloc)))

            logger.verbose("Processing in %s threads", self.batch_size)

            self.init.set()
            logger.info("Initialized S3FD Detector.")
        except Exception as err:
            self.error.set()
            raise err

    def detect_faces(self, *args, **kwargs):
        """ Detect faces in Multiple Threads """
        super().detect_faces(*args, **kwargs)
        workers = MultiThread(target=self.detect_thread, thread_count=self.batch_size)
        workers.start()
        workers.join()
        sentinel = self.queues["in"].get()
        self.queues["out"].put(sentinel)
        logger.debug("Detecting Faces complete")

    def detect_thread(self):
        """ Detect faces in rgb image """
        logger.debug("Launching Detect")
        while True:
            item = self.get_item()
            if item == "EOF":
                break
            logger.trace("Detecting faces: '%s'", item["filename"])
            detect_image, scale = self.compile_detection_image(item["image"], is_square=True)
            for angle in self.rotation:
                current_image, rotmat = self.rotate_image(detect_image, angle)
                faces = self.model.detect_face(current_image)
                if angle != 0 and faces.any():
                    logger.verbose("found face(s) by rotating image %s degrees", angle)
                if faces.any():
                    break

            detected_faces = self.process_output(faces, rotmat, scale)
            item["detected_faces"] = detected_faces
            self.finalize(item)

        logger.debug("Thread Completed Detect")

    def process_output(self, faces, rotation_matrix, scale):
        """ Compile found faces for output """
        logger.trace("Processing Output: (faces: %s, rotation_matrix: %s)", faces, rotation_matrix)
        faces = [self.to_bounding_box_dict(face[0], face[1], face[2], face[3]) for face in faces]
        if isinstance(rotation_matrix, np.ndarray):
            faces = [self.rotate_rect(face, rotation_matrix)
                     for face in faces]
        detected = [self.to_bounding_box_dict(face["left"] / scale, face["top"] / scale,
                                              face["right"] / scale, face["bottom"] / scale)
                    for face in faces]
        logger.trace("Processed Output: %s", detected)
        return detected


class S3fd():
    """ Tensorflow Network """
    def __init__(self, model_path, target_size, vram_ratio, card_id, confidence):
        logger.debug("Initializing: %s: (model_path: '%s', target_size: %s, vram_ratio: %s, "
                     "card_id: %s)",
                     self.__class__.__name__, model_path, target_size, vram_ratio, card_id)
        # Must import tensorflow inside the spawned process for Windows machines
        import tensorflow as tf
        self.is_gpu = False
        self.tf = tf  # pylint: disable=invalid-name
        self.model_path = model_path
 </s> add class O2K_Sum(O2K_ReduceLayer):
    def call(self, x, *args):
        return K.sum(x, self._axes, self._keepdims)


class O2K_Sqrt(keras.engine.Layer):
    def call(self, x, *args):
        return K.sqrt(x)


class O2K_Pow(keras.engine.Layer):
    def call(self, x, *args):
        return K.pow(*x)


class O2K_ConstantLayer(keras.engine.Layer):
    def __init__(self, constant_obj, dtype, **kwargs):
        self._dtype = np.dtype(dtype).name
        self._constant = np.array(constant_obj, dtype=self._dtype)
        super(O2K_ConstantLayer, self).__init__(**kwargs)

    def call(self, *args):
        # pylint:disable=arguments-differ
        data = K.constant(self._constant, dtype=self._dtype)
        return data

    def compute_output_shape(self, input_shape):
        return self._constant.shape

    def get_config(self):
        config = super(O2K_ConstantLayer, self).get_config()
        config.update({
            'constant_obj': self._constant,
            'dtype': self._dtype
        })
        return config


class O2K_Div(O2K_ElementwiseLayer):
    # pylint:disable=arguments-differ
    def call(self, x, *args):
        return x[0] / x[1]


class S3fd(KSession):
    """ Keras Network """
    def __init__(self, model_path, model_kwargs, confidence):
        logger.debug("Initializing: %s: (model_path: '%s')",
                     self.__class__.__name__, model_path)
        super().__init__("S3FD", model_path, model_kwargs)
        self.load_model()
    return cpu_count()
  
  
  def total_cpus():
      """ Return total number of cpus """
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
    return cpu_count()
  
  
  class FSThread(threading.Thread):
      """ Subclass of thread that passes errors back to parent """
      def __init__(self, group=None, target=None, name=None,  # pylint: disable=too-many-arguments
keep keep keep keep replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace replace keep keep keep keep keep
 De-Multiprocess Extract (#871)

* requirements.txt: - Pin opencv to 4.1.1 (fixes cv2-dnn error)

* lib.face_detect.DetectedFace: change LandmarksXY to landmarks_xy. Add left, right, top, bottom attributes

* lib.model.session: Session manager for loading models into different graphs (for Nvidia + CPU)

* plugins.extract._base: New parent class for all extract plugins

* plugins.extract.pipeline. Remove MultiProcessing. Dynamically limit batchsize for Nvidia cards. Remove loglevel input

* S3FD + FAN plugins. Standardise to Keras version for all backends

* Standardize all extract plugins to new threaded codebase

* Documentation. Start implementing Numpy style docstrings for Sphinx Documentation

* Remove s3fd_amd. Change convert OTF to expect DetectedFace object

* faces_detect - clean up and documentation

* Remove PoolProcess

* Migrate manual tool to new extract workflow

* Remove AMD specific extractor code from cli and plugins

* Sort tool to new extract workflow

* Remove multiprocessing from project

* Remove multiprocessing queues from QueueManager

* Remove multiprocessing support from logger

* Move face_filter to new extraction pipeline

* Alignments landmarksXY > landmarks_xy and legacy handling

* Intercept get_backend for sphinx doc build

# Add Sphinx documentation 
lib/multithreading.py