          FaceswapError
              If the given location is a file and does not have a valid video extension.
  
          """
          if os.path.isdir(self.location):
              retval = False
          elif os.path.splitext(self.location)[1].lower() in _video_extensions:
              retval = True
          else:
              raise FaceswapError("The input file '{}' is not a valid video".format(self.location))
 
</s> remove         if self._skip_num == 1 and not self._alignments.data:
            logger.debug("No frames to be skipped")
 </s> add         error = False
        retval: List[List[str]] = []
        for files in (filter_files, nfilter_files):
            filt_files = [files] if isinstance(files, str) else files
            filt_files = [] if filt_files is None else filt_files
            for file in filt_files:
                if (not os.path.isfile(file) or
                        os.path.splitext(file)[-1].lower() not in _image_extensions):
                    logger.warning("Filter file '%s' does not exist or is not an image file", file)
                    error = True
            retval.append(filt_files)

        filters = retval[0]
        nfilters = retval[1]
        f_fnames = set(os.path.basename(fname) for fname in filters)
        n_fnames = set(os.path.basename(fname) for fname in nfilters)
        if f_fnames.intersection(n_fnames):
            error = True
            logger.warning("filter and nfilter filenames should be unique. The following "
                           "filenames exist in both folders: %s", f_fnames.intersection(n_fnames))

        if error:
            logger.error("There was a problem processing filter files. See the above warnings for "
                         "details")
            sys.exit(1)
        logger.debug("filter_files: %s, nfilter_files: %s", retval[0], retval[1])

        return filters, nfilters

    @classmethod
    def _identity_from_extracted(cls, filename) -> Tuple[np.ndarray, bool]:
        """ Test whether the given image is a faceswap extracted face and contains identity
        information. If so, return the identity embedding

        Parameters
        ----------
        filename: str
            Full path to the image file to load

        Returns
        -------
        :class:`numpy.ndarray`
            The identity embeddings, if they can be obtained from the image header, otherwise an
            empty array
        bool
            ``True`` if the image is a faceswap extracted image otherwise ``False``
        """
        if os.path.splitext(filename)[-1].lower() != ".png":
            logger.info("'%s' not a png. Returning empty array", filename)
            return np.array([]), False

        meta = read_image_meta(filename)
        if "itxt" not in meta or "alignments" not in meta["itxt"]:
            logger.debug("'%s' does not contain faceswap data. Returning empty array", filename)
            return np.array([]), False

        align: "PNGHeaderAlignmentsDict" = meta["itxt"]["alignments"]
        if "identity" not in align or "vggface2" not in align["identity"]:
            logger.debug("'%s' does not contain identity data. Returning empty array", filename)
            return np.array([]), True

        retval = np.array(align["identity"]["vggface2"])
        logger.debug("Obtained identity for '%s'. Shape: %s", filename, retval.shape)

        return retval, True

    def _process_extracted(self, item: ExtractMedia) -> None:
        """ Process the output from the extraction pipeline.

        If no face has been detected, or multiple faces are detected for the inclusive filter,
        embeddings and filenames are removed from the filter.

        if a single face is detected or multiple faces are detected for the exclusive filter,
        embeddings are added to the relevent filter list

        Parameters
        ----------
        item: :class:`plugins.extract.Pipeline.ExtracMedia`
            The output from the extraction pipeline containing the identity encodings
        """
        is_filter = item.filename in self._filter_files
        lbl = "filter" if is_filter else "nfilter"
        filelist = self._filter_files if is_filter else self._nfilter_files
        embeddings = self._embeddings if is_filter else self._nembeddings
        identities = np.array([face.identity["vggface2"] for face in item.detected_faces])
        idx = filelist.index(item.filename)

        if len(item.detected_faces) == 0:
            logger.warning("No faces detected for %s in file '%s'. Image will not be used",
                           lbl, os.path.basename(item.filename))
            filelist.pop(idx)
            embeddings.pop(idx) </s> remove         retval = np.frombuffer(png_write_meta(retval.tobytes(), metadata), dtype="uint8")
 </s> add         retval = png_write_meta(retval.tobytes(), metadata) </s> remove         # from lib.queue_manager import queue_manager ; queue_manager.debug_monitor(3)
 </s> add         logger.info("Extracting faces to obtain identity from images")
        logger.debug("Files requiring full extraction: %s",
                     [fname for fname in file_list if fname not in aligned])
        logger.debug("Aligned files requiring identity info: %s", aligned)

        loader = PipelineLoader(file_list, self._extractor, aligned_filenames=aligned)
        loader.launch()

        for phase in range(self._extractor.passes):
            is_final = self._extractor.final_pass
            detected_faces: Dict[str, ExtractMedia] = {}
            self._extractor.launch()
            desc = "Obtaining reference face Identity"
            if self._extractor.passes > 1:
                desc = (f"{desc } pass {phase + 1} of {self._extractor.passes}: "
                        f"{self._extractor.phase_text}")
            for extract_media in tqdm(self._extractor.detected_faces(),
                                      total=len(file_list),
                                      file=sys.stdout,
                                      desc=desc):
                if is_final:
                    self._process_extracted(extract_media)
                else:
                    extract_media.remove_image()
                    # cache extract_media for next run
                    detected_faces[extract_media.filename] = extract_media

            if not is_final:
                logger.debug("Reloading images")
                loader.reload(detected_faces)

        self._extractor.reset_phase_index()

    def _get_embeddings(self) -> None:
        """ Obtain the embeddings for the given filter lists """
        needs_extraction: List[str] = []
        aligned: List[str] = []

        for files, embed in zip((self._filter_files, self._nfilter_files),
                                (self._embeddings, self._nembeddings)):
            for idx, file in enumerate(files):
                identity, is_aligned = self._identity_from_extracted(file)
                if np.any(identity):
                    logger.debug("Obtained identity from png header: '%s'", file)
                    embed[idx] = identity[None, ...]
                    continue

                needs_extraction.append(file)
                if is_aligned:
                    aligned.append(file)

        if needs_extraction:
            self._identity_from_extractor(needs_extraction, aligned)

        if not self._nfilter_files and not self._filter_files:
            logger.error("No faces were detected from your selected identity filter files")
            sys.exit(1)

        logger.debug("Filter: (filenames: %s, shape: %s), nFilter: (filenames: %s, shape: %s)",
                     [os.path.basename(f) for f in self._filter_files],
                     self.embeddings.shape,
                     [os.path.basename(f) for f in self._nfilter_files],
                     self.n_embeddings.shape)


class PipelineLoader():
    """ Handles loading and reloading images into the extraction pipeline.

    Parameters
    ----------
    path: str or list of str
        Full path to a folder of images or a video file or a list of image files
    extractor: :class:`~plugins.extract.pipeline.Extractor`
        The extractor pipeline for obtaining face identity from images
    aligned_filenames: list, optional
        Used for when the loader is used for getting face filter embeddings. List of full path to
        image files that exist in :attr:`path` that are aligned faceswap images
    """
    def __init__(self,
                 path: Union[str, List[str]],
                 extractor: Extractor,
                 aligned_filenames: Optional[List[str]] = None) -> None:
        logger.debug("Initializing %s: (path: %s, extractor: %s, aligned_filenames: %s)",
                     self.__class__.__name__, path, extractor, aligned_filenames)
        self._images = ImagesLoader(path, fast_count=True)
        self._extractor = extractor
        self._threads: List[MultiThread] = []
        self._aligned_filenames = [] if aligned_filenames is None else aligned_filenames
        logger.debug("Initialized %s", self.__class__.__name__)

    @property
    def is_video(self) -> bool:
        """ bool: ``True`` if the input location is a video file, ``False`` if it is a folder of
        images """
        return self._images.is_video

    @property
    def file_list(self) -> List[str]:
        """ list: A full list of files in the source location. If the input is a video
        then this is a list of dummy filenames as corresponding to an alignments file """
        return self._images.file_list

    @property
    def process_count(self) -> int:
        """ int: The number of images or video frames to be processed (IE the total count less
        items that are to be skipped from the :attr:`skip_list`)"""
        return self._images.process_count

    def add_skip_list(self, skip_list: List[int]) -> None:
        """ Add a skip list to the :class:`ImagesLoader`

        Parameters
        ----------
        skip_list: list
            A list of indices corresponding to the frame indices that should be skipped by the
            :func:`load` function.
        """
        self._images.add_skip_list(skip_list)

    def launch(self) -> None:
        """ Launch the image loading pipeline """ </s> remove     def _set_skip_list(self) -> None:
        """ Add the skip list to the image loader
 </s> add     @property
    def n_embeddings(self) -> np.ndarray:
        """ :class:`numpy.ndarray`: The n-filter embeddings"""
        if self._nembeddings and all(np.any(e) for e in self._nembeddings):
            retval = np.concatenate(self._nembeddings, axis=0)
        else:
            retval = np.array([])
        return retval
        if not isinstance(self.location, str) or os.path.isdir(self.location):
          FaceswapError
              If the given location is a file and does not have a valid video extension.
  
          """
        if not isinstance(self.location, str) or os.path.isdir(self.location):
              retval = False
          elif os.path.splitext(self.location)[1].lower() in _video_extensions:
              retval = True
          else:
              raise FaceswapError("The input file '{}' is not a valid video".format(self.location))
keep keep keep keep replace keep keep keep keep keep
 Update Face Filter
  - Remove old face filter
  - plugins.extract.pipeline: Expose plugins directly
  - Change `is_aligned` from plugin level to ExtractMedia level
  - Allow extract pipeline to take faceswap aligned images
  - Add ability for recognition plugins to accept aligned faces as input
  - Add face filter to recognition plugin
  - Move extractor pipeline IO ops to own class 
lib/image.py