Module facetorch.analyzer.core

Classes

class FaceAnalyzer (cfg: omegaconf.omegaconf.OmegaConf)

Expand source code

class FaceAnalyzer(object):
    @Timer(
        "FaceAnalyzer.__init__", "{name}: {milliseconds:.2f} ms", logger=logger.debug
    )
    def __init__(self, cfg: OmegaConf):
        """FaceAnalyzer is the main class that reads images, runs face detection, tensor unification and facial feature prediction.
        It also draws bounding boxes and facial landmarks over the image.

        The following components are used:

        1. Reader - reads the image and returns an ImageData object containing the image tensor.
        2. Detector - wrapper around a neural network that detects faces.
        3. Unifier - processor that unifies sizes of all faces and normalizes them between 0 and 1.
        4. Predictor dict - dict of wrappers around neural networks trained to analyze facial features.
        5. Utilizer dict - dict of utilizer processors that can for example extract 3D face landmarks or draw boxes over the image.

        Args:
            cfg (OmegaConf): Config object with image reader, face detector, unifier and predictor configurations.

        Attributes:
            cfg (OmegaConf): Config object with image reader, face detector, unifier and predictor configurations.
            reader (BaseReader): Reader object that reads the image and returns an ImageData object containing the image tensor.
            detector (FaceDetector): Lazily loaded and cached FaceDetector object.
            unifier (FaceUnifier): FaceUnifier object that unifies sizes of all faces and normalizes them between 0 and 1.
            predictors (MutableMapping[str, FacePredictor]): Mapping of lazily loaded
                and cached predictors. Iterating names does not load models; accessing
                a value does.
            utilizers (MutableMapping[str, FaceUtilizer]): Mapping of lazily loaded
                utilizer objects. Selection-linked utilizers load only when their
                predictor ran.
            logger (logging.Logger): Logger object that logs messages to the console or to a file.

        """
        self.cfg = cfg
        self._component_lock = threading.RLock()

        if hasattr(self.cfg, "logger") and self.cfg.logger is not None:
            self.logger = instantiate(self.cfg.logger).logger
        else:
            self.logger = LoggerJsonFile(level=logging.INFO).logger

        self.logger.info("Initializing FaceAnalyzer")
        self.logger.debug("Config", extra=self.cfg.__dict__["_content"])

        self.logger.info("Initializing BaseReader")
        self.reader = instantiate(self.cfg.reader)
        self._reader_signature_owner = None
        self._reader_signature_parameters = None

        self.logger.info("Registering lazy FaceDetector")
        self._detector_config = self.cfg.detector if "detector" in self.cfg else None
        self._detector = _UNLOADED

        self.logger.info("Initializing FaceUnifier")
        if "unifier" in self.cfg:
            self.unifier = instantiate(self.cfg.unifier)
        else:
            self.unifier = None

        self.logger.info("Registering lazy FacePredictor objects")
        predictor_configs = {}
        if "predictor" in self.cfg:
            for predictor_name in self.cfg.predictor:
                self.logger.info(f"Registering FacePredictor {predictor_name}")
                predictor_configs[predictor_name] = self.cfg.predictor[predictor_name]
        self._predictors = _LazyComponentRegistry(
            predictor_configs,
            loader=self._load_predictor,
            lock=self._component_lock,
        )

        utilizer_configs = {}
        if "utilizer" in self.cfg:
            self.logger.info("Registering lazy BaseUtilizer objects")
            for utilizer_name in self.cfg.utilizer:
                self.logger.info(f"Registering BaseUtilizer {utilizer_name}")
                utilizer_configs[utilizer_name] = self.cfg.utilizer[utilizer_name]
        self._utilizers = _LazyComponentRegistry(
            utilizer_configs,
            loader=self._load_utilizer,
            lock=self._component_lock,
        )
        dependencies = (
            self.cfg.utilizer_dependencies
            if "utilizer_dependencies" in self.cfg
            else None
        )
        self._utilizer_dependencies = self._normalize_utilizer_dependencies(
            dependencies,
            utilizer_names=tuple(utilizer_configs),
            predictor_names=tuple(predictor_configs),
        )

    def __call__(self, *args, **kwargs):
        return self.run(*args, **kwargs)

    @property
    def detector(self):
        """Return the configured detector, constructing and caching it on demand."""
        detector = self.__dict__.get("_detector", _UNLOADED)
        if detector is not _UNLOADED:
            return detector

        detector_config = self.__dict__.get("_detector_config")
        if detector_config is None:
            raise ConfigurationError("No face detector is configured.")

        lock = self._get_component_lock()
        with lock:
            if self._detector is _UNLOADED:
                self.logger.info("Initializing FaceDetector")
                self._detector = instantiate(detector_config)
        return self._detector

    @detector.setter
    def detector(self, detector) -> None:
        """Install an already-constructed detector, primarily for extensions/tests."""
        self._detector_config = None
        self._detector = detector

    @property
    def predictors(self) -> MutableMapping[str, FacePredictor]:
        """Return the lazy predictor mapping without constructing its values."""
        registry = self.__dict__.get("_predictors")
        if registry is None:
            registry = _LazyComponentRegistry(lock=self._get_component_lock())
            self._predictors = registry
        return registry

    @predictors.setter
    def predictors(self, predictors: Mapping[str, FacePredictor]) -> None:
        """Replace configured predictors with already-constructed components."""
        if not isinstance(predictors, Mapping):
            raise TypeError("predictors must be a mapping from names to predictors.")
        self._predictors = _LazyComponentRegistry(
            loaded=predictors,
            lock=self._get_component_lock(),
        )

    @property
    def configured_predictors(self) -> tuple[str, ...]:
        """Predictor names in deterministic configuration order, without loading."""
        return tuple(self.predictors)

    @property
    def loaded_predictors(self) -> tuple[str, ...]:
        """Predictor names whose wrappers and models are already cached."""
        registry = self.predictors
        if isinstance(registry, _LazyComponentRegistry):
            return registry.loaded_names
        return tuple(registry)

    @property
    def utilizers(self) -> MutableMapping[str, Any]:
        """Return the lazy utilizer mapping without constructing its values."""
        registry = self.__dict__.get("_utilizers")
        if registry is None:
            registry = _LazyComponentRegistry(lock=self._get_component_lock())
            self._utilizers = registry
        return registry

    @utilizers.setter
    def utilizers(self, utilizers: Mapping[str, Any]) -> None:
        """Replace configured utilizers with already-constructed components."""
        if not isinstance(utilizers, Mapping):
            raise TypeError("utilizers must be a mapping from names to utilizers.")
        self._utilizers = _LazyComponentRegistry(
            loaded=utilizers,
            lock=self._get_component_lock(),
        )

    @property
    def configured_utilizers(self) -> tuple[str, ...]:
        """Utilizer names in deterministic configuration order, without loading."""
        return tuple(self.utilizers)

    @property
    def loaded_utilizers(self) -> tuple[str, ...]:
        """Utilizer names whose objects are already cached."""
        registry = self.utilizers
        if isinstance(registry, _LazyComponentRegistry):
            return registry.loaded_names
        return tuple(registry)

    @property
    def utilizer_dependencies(self) -> dict[str, tuple[str, ...]]:
        """Explicit predictor requirements for configured utilizers."""
        return dict(self.__dict__.get("_utilizer_dependencies", {}))

    @property
    def detector_loaded(self) -> bool:
        """Whether the detector wrapper and model are already cached."""
        return self.__dict__.get("_detector", _UNLOADED) is not _UNLOADED

    def _get_component_lock(self):
        lock = self.__dict__.get("_component_lock")
        if lock is None:
            lock = threading.RLock()
            self._component_lock = lock
        return lock

    def _load_predictor(self, name: str, predictor_config) -> FacePredictor:
        self.logger.info(f"Initializing FacePredictor {name}")
        return instantiate(predictor_config)

    def _load_utilizer(self, name: str, utilizer_config):
        self.logger.info(f"Initializing BaseUtilizer {name}")
        return instantiate(utilizer_config)

    @staticmethod
    def _normalize_predictor_selection(
        selection: Optional[Iterable[str]], option_name: str
    ) -> Optional[tuple[str, ...]]:
        if selection is None:
            return None
        if isinstance(selection, (str, bytes)):
            raise ConfigurationError(
                f"{option_name} must be a collection of predictor names, not a string."
            )
        try:
            names = tuple(selection)
        except TypeError as exc:
            raise ConfigurationError(
                f"{option_name} must be a collection of predictor names."
            ) from exc

        invalid = [name for name in names if not isinstance(name, str) or not name]
        if invalid:
            raise ConfigurationError(
                f"{option_name} must contain only non-empty predictor names."
            )

        seen = set()
        duplicates = []
        for name in names:
            if name in seen and name not in duplicates:
                duplicates.append(name)
            seen.add(name)
        if duplicates:
            raise ConfigurationError(
                f"{option_name} contains duplicate predictor names: "
                + ", ".join(duplicates)
                + "."
            )
        return names

    @classmethod
    def _normalize_utilizer_dependencies(
        cls,
        dependencies: Optional[Mapping[str, Iterable[str]]],
        *,
        utilizer_names: tuple[str, ...],
        predictor_names: tuple[str, ...],
    ) -> dict[str, tuple[str, ...]]:
        """Validate the explicit utilizer-to-predictor execution graph."""
        if dependencies is None:
            return {}
        if not isinstance(dependencies, Mapping):
            raise ConfigurationError(
                "utilizer_dependencies must map utilizer names to predictor names."
            )

        configured_utilizers = set(utilizer_names)
        configured_predictors = set(predictor_names)
        normalized = {}
        for utilizer_name, requirements in dependencies.items():
            if not isinstance(utilizer_name, str) or not utilizer_name:
                raise ConfigurationError(
                    "utilizer_dependencies must use non-empty utilizer names."
                )
            if utilizer_name not in configured_utilizers:
                raise ConfigurationError(
                    f"Unknown utilizer dependency target {utilizer_name!r}. "
                    "Configured utilizers: "
                    + (", ".join(utilizer_names) if utilizer_names else "none")
                    + "."
                )

            option_name = f"utilizer_dependencies[{utilizer_name!r}]"
            requirement_names = cls._normalize_predictor_selection(
                requirements, option_name
            )
            if requirement_names is None:
                raise ConfigurationError(
                    f"{option_name} must be a collection of predictor names."
                )
            unknown = [
                name for name in requirement_names if name not in configured_predictors
            ]
            if unknown:
                raise ConfigurationError(
                    f"{option_name} references unknown predictor name(s): "
                    + ", ".join(unknown)
                    + ". Configured predictors: "
                    + (", ".join(predictor_names) if predictor_names else "none")
                    + "."
                )
            normalized[utilizer_name] = requirement_names
        return normalized

    def _select_predictor_names(
        self,
        include_predictors: Optional[Iterable[str]],
        exclude_predictors: Optional[Iterable[str]],
    ) -> tuple[str, ...]:
        """Validate selection and return names in configuration order."""
        include = self._normalize_predictor_selection(
            include_predictors, "include_predictors"
        )
        exclude = self._normalize_predictor_selection(
            exclude_predictors, "exclude_predictors"
        )
        if include is not None and exclude is not None:
            raise ConfigurationError(
                "Cannot specify both include_predictors and exclude_predictors. "
                "Use one or the other."
            )

        configured = self.configured_predictors
        configured_set = set(configured)
        requested = include if include is not None else exclude
        unknown = (
            [name for name in requested if name not in configured_set]
            if requested is not None
            else []
        )
        if unknown:
            raise ConfigurationError(
                "Unknown predictor name(s): "
                + ", ".join(unknown)
                + ". Configured predictors: "
                + (", ".join(configured) if configured else "none")
                + "."
            )

        if include is not None:
            included = set(include)
            return tuple(name for name in configured if name in included)
        if exclude is not None:
            excluded = set(exclude)
            return tuple(name for name in configured if name not in excluded)
        return configured

    @Timer("FaceAnalyzer.run", "{name}: {milliseconds:.2f} ms", logger=logger.debug)
    def run(
        self,
        image_source: Optional[
            Union[str, os.PathLike, torch.Tensor, np.ndarray, bytes, Image.Image]
        ] = None,
        path_image: Optional[str] = None,
        face_batch_size: Optional[int] = None,
        fix_img_size: bool = False,
        return_img_data: Optional[bool] = None,
        include_tensors: bool = False,
        path_output: Optional[str] = None,
        tensor: Optional[torch.Tensor] = None,
        include_predictors: Optional[List[str]] = None,
        exclude_predictors: Optional[List[str]] = None,
        skip_detector: bool = False,
        *,
        batch_size: Optional[int] = None,
        input_policy: str = "coerce",
        input_spec: Optional[InputSpec] = None,
    ) -> AnalysisResult:
        """Analyze exactly one source image and return one stable result type.

        Args:
            image_source: Input accepted by the configured reader. The default
                reader accepts local paths, tensors, NumPy arrays, bytes, and PIL
                images. URLs require an explicit URLReader configuration.
            path_image (Optional[str]): Deprecated. Use image_source instead.
            face_batch_size (Optional[int]): Number of faces from this image sent to
                each predictor at once. This is an upper bound; predictors may use
                smaller chunks to honor their model artifact. Default: 8.
            fix_img_size (bool): If True, resizes the image to the size specified in reader. Default is False.
            return_img_data (Optional[bool]): Deprecated no-op. Use
                ``include_tensors`` and the fields on ``AnalysisResult`` or call
                ``run_legacy`` for the former flag-dependent return type.
            include_tensors (bool): If True, includes tensors in the returned data object. If False, tensors are removed. Default is False.
            path_output (Optional[str]): Path where to save the image with detected faces. If None, the image is not saved. Default: None.
            tensor (Optional[torch.Tensor]): Deprecated. Use image_source instead.
            include_predictors (Optional[List[str]]): Names to run. None runs all
                configured predictors and an empty collection runs none.
            exclude_predictors (Optional[List[str]]): Names to omit. None and an
                empty collection omit none. Cannot be combined with an include.
            skip_detector (bool): If True, skip face detection, avoid constructing
                its model, and treat the input as a pre-cropped face. Default: False.
            batch_size (Optional[int]): Deprecated warning alias for
                ``face_batch_size`` throughout v1.x.
            input_policy (str): ``coerce`` (default) or ``strict``.
            input_spec (Optional[InputSpec]): Explicit source layout/range/color
                description, especially for strict-mode conversions.

        Returns:
            AnalysisResult: Stable result for the one source image.

        """

        compatibility_warnings = []

        if face_batch_size is not None and batch_size is not None:
            raise ConfigurationError(
                "Specify only face_batch_size; batch_size is its deprecated alias."
            )
        if batch_size is not None:
            message = (
                "batch_size is deprecated and will be removed after v1.x; "
                "use face_batch_size."
            )
            warnings.warn(message, DeprecationWarning, stacklevel=2)
            compatibility_warnings.append(message)
            effective_face_batch_size = batch_size
        elif face_batch_size is None:
            effective_face_batch_size = 8
        else:
            effective_face_batch_size = face_batch_size

        if (
            isinstance(effective_face_batch_size, bool)
            or not isinstance(effective_face_batch_size, int)
            or effective_face_batch_size < 1
        ):
            raise ConfigurationError(
                "face_batch_size must be an integer greater than or equal to 1, "
                f"got {effective_face_batch_size!r}."
            )

        if return_img_data is not None:
            message = (
                "return_img_data no longer changes FaceAnalyzer.run's return type; "
                "use include_tensors or the explicit run_legacy adapter."
            )
            warnings.warn(message, DeprecationWarning, stacklevel=2)
            compatibility_warnings.append(message)

        def _run_component(label, operation):
            try:
                return operation()
            except FacetorchError:
                raise
            except Exception as exc:
                raise InferenceError(f"{label} failed during analysis.") from exc

        def _predict_batch(
            data: ImageData, predictor: FacePredictor, predictor_name: str
        ) -> ImageData:
            n_faces = len(data.faces)
            predictor_limit = getattr(predictor, "max_batch_size", None)
            if predictor_limit is not None and (
                isinstance(predictor_limit, bool)
                or not isinstance(predictor_limit, int)
                or predictor_limit < 1
            ):
                raise ConfigurationError(
                    f"Face predictor {predictor_name!r} has invalid "
                    f"max_batch_size {predictor_limit!r}."
                )
            chunk_size = (
                effective_face_batch_size
                if predictor_limit is None
                else min(effective_face_batch_size, predictor_limit)
            )

            for face_indx_start in range(0, n_faces, chunk_size):
                face_indx_end = min(face_indx_start + chunk_size, n_faces)

                face_batch_tensor = torch.stack(
                    [face.tensor for face in data.faces[face_indx_start:face_indx_end]]
                )
                preds = predictor.run(face_batch_tensor)
                expected_count = face_indx_end - face_indx_start
                try:
                    prediction_count = len(preds)
                except TypeError as exc:
                    raise InferenceError(
                        f"Face predictor {predictor_name!r} must return one "
                        "prediction per input face; its result has no length."
                    ) from exc
                if prediction_count != expected_count:
                    raise InferenceError(
                        f"Face predictor {predictor_name!r} returned "
                        f"{prediction_count} prediction(s) for "
                        f"{expected_count} input face(s)."
                    )
                data.add_preds(preds, predictor_name, face_indx_start)

            return data

        self.logger.info("Running FaceAnalyzer")
        selected_predictors = self._select_predictor_names(
            include_predictors, exclude_predictors
        )
        configured_predictors = set(self.configured_predictors)
        if skip_detector and selected_predictors and self.unifier is None:
            raise ConfigurationError(
                "skip_detector=True with selected predictors requires a face "
                "unifier. Configure analyzer.unifier or pass "
                "include_predictors=[] for predictor-free processing."
            )

        supplied_sources = [
            name
            for name, value in (
                ("image_source", image_source),
                ("path_image", path_image),
                ("tensor", tensor),
            )
            if value is not None
        ]
        if len(supplied_sources) > 1:
            raise InputError(
                "Supply exactly one input source using image_source. Received: "
                + ", ".join(supplied_sources)
                + "."
            )
        if not supplied_sources:
            raise InputError(
                "image_source is required. Pass a file path, URL, tensor, numpy array, "
                "bytes, or PIL Image."
            )
        if path_image is not None:
            message = "path_image is deprecated; use image_source."
            warnings.warn(message, DeprecationWarning, stacklevel=2)
            compatibility_warnings.append(message)
            image_source = path_image
        elif tensor is not None:
            message = "tensor is deprecated; use image_source."
            warnings.warn(message, DeprecationWarning, stacklevel=2)
            compatibility_warnings.append(message)
            image_source = tensor

        self.logger.info("Reading image")
        data = self._read_input(
            image_source,
            fix_img_size,
            input_policy=input_policy,
            input_spec=input_spec,
        )
        data.warnings.extend(compatibility_warnings)

        path_output = None if path_output == "None" else path_output
        data.path_output = path_output

        try:
            data.version = version("facetorch")
        except Exception as e:
            self.logger.warning("Could not get version number", extra={"error": e})

        if skip_detector:
            self.logger.info("Skipping detector (skip_detector=True)")
            face_tensor = data.tensor[0]
            face = Face(
                indx=0,
                loc=Location(
                    x1=0, y1=0, x2=data.dims.width, y2=data.dims.height
                ),
                dims=Dimensions(
                    height=data.dims.height, width=data.dims.width
                ),
                tensor=face_tensor,
                ratio=1.0,
            )
            data.faces = [face]
            n_faces = 1
        else:
            self.logger.info("Detecting faces")
            data = _run_component("Face detector", lambda: self.detector.run(data))
            n_faces = len(data.faces)

        self.logger.info(f"Number of faces: {n_faces}")

        ran_predictors = set()
        if n_faces > 0:
            if selected_predictors and self.unifier is None:
                raise ConfigurationError(
                    "Detected faces cannot be sent to selected predictors without "
                    "a face unifier. Configure analyzer.unifier or pass "
                    "include_predictors=[] for predictor-free processing."
                )

            if self.unifier is not None:
                self.logger.info("Unifying faces")
                data = _run_component("Face unifier", lambda: self.unifier.run(data))

            self.logger.info("Predicting facial features")
            for predictor_name in selected_predictors:
                self.logger.info(f"Running FacePredictor: {predictor_name}")
                data = _run_component(
                    f"Face predictor {predictor_name!r}",
                    lambda predictor_name=predictor_name: _predict_batch(
                        data,
                        self.predictors[predictor_name],
                        predictor_name,
                    ),
                )
                ran_predictors.add(predictor_name)

            utilizer_names = tuple(self.utilizers)
        else:
            utilizer_names = ("save",) if "save" in self.utilizers else ()

        self.logger.info("Utilizing facial features")
        dependencies = self.__dict__.get("_utilizer_dependencies", {})
        for utilizer_name in utilizer_names:
            required_predictors = set(dependencies.get(utilizer_name, ()))
            unknown_requirements = required_predictors - configured_predictors
            if unknown_requirements:
                raise ConfigurationError(
                    f"Utilizer {utilizer_name!r} requires unknown predictor(s): "
                    + ", ".join(sorted(unknown_requirements))
                    + "."
                )
            missing_predictors = required_predictors - ran_predictors
            if missing_predictors:
                self.logger.info(
                    f"Skipping BaseUtilizer: {utilizer_name} "
                    "(required predictor(s) not run: "
                    + ", ".join(sorted(missing_predictors))
                    + ")"
                )
                continue
            self.logger.info(f"Running BaseUtilizer: {utilizer_name}")
            data = _run_component(
                f"Face utilizer {utilizer_name!r}",
                lambda utilizer_name=utilizer_name: self.utilizers[
                    utilizer_name
                ].run(data),
            )

        if not include_tensors:
            self.logger.debug(
                "Removing tensors from response as include_tensors is False"
            )
            data.reset_tensors()

        result = AnalysisResult.from_image_data(data, include_tensors=include_tensors)
        self.logger.debug(
            "Returning analysis result",
            extra={"face_count": len(result.faces), "version": result.version},
        )
        return result

    def run_legacy(
        self,
        image_source: Optional[
            Union[str, os.PathLike, torch.Tensor, np.ndarray, bytes, Image.Image]
        ] = None,
        *legacy_args: Any,
        path_image: Optional[str] = None,
        batch_size: int = 8,
        fix_img_size: bool = False,
        return_img_data: bool = False,
        include_tensors: bool = False,
        path_output: Optional[str] = None,
        tensor: Optional[torch.Tensor] = None,
        include_predictors: Optional[List[str]] = None,
        exclude_predictors: Optional[List[str]] = None,
        skip_detector: bool = False,
        input_policy: str = "coerce",
        input_spec: Optional[InputSpec] = None,
    ) -> Union[Response, ImageData]:
        """Run the canonical pipeline using either shipped v0 positional order.

        Releases through v0.4 used ``path_image, batch_size, ...``. Releases
        v0.5 and v0.6 inserted ``image_source, path_image, batch_size, ...``.
        This adapter accepts both layouts and preserves their flag-dependent
        return type. Prefer ``run()`` for new integrations.
        """
        warnings.warn(
            "FaceAnalyzer.run_legacy is a v1.x compatibility adapter; migrate to run().",
            DeprecationWarning,
            stacklevel=2,
        )

        if legacy_args:
            # The second positional value distinguishes the two historical
            # layouts: v0.5+ path_image was a path/None, while earlier releases
            # placed the integer batch_size there.
            has_v05_source_slot = legacy_args[0] is None or isinstance(
                legacy_args[0], (str, os.PathLike)
            )
            positional_names = (
                (
                    "path_image",
                    "batch_size",
                    "fix_img_size",
                    "return_img_data",
                    "include_tensors",
                    "path_output",
                    "tensor",
                    "include_predictors",
                    "exclude_predictors",
                    "skip_detector",
                )
                if has_v05_source_slot
                else (
                    "batch_size",
                    "fix_img_size",
                    "return_img_data",
                    "include_tensors",
                    "path_output",
                    "tensor",
                    "include_predictors",
                    "exclude_predictors",
                    "skip_detector",
                )
            )
            if len(legacy_args) > len(positional_names):
                maximum = len(positional_names) + 1
                supplied = len(legacy_args) + 1
                raise TypeError(
                    "run_legacy() takes at most "
                    f"{maximum} positional arguments but {supplied} were given"
                )

            legacy_values = {
                "path_image": path_image,
                "batch_size": batch_size,
                "fix_img_size": fix_img_size,
                "return_img_data": return_img_data,
                "include_tensors": include_tensors,
                "path_output": path_output,
                "tensor": tensor,
                "include_predictors": include_predictors,
                "exclude_predictors": exclude_predictors,
                "skip_detector": skip_detector,
            }
            default_values = {
                "path_image": None,
                "batch_size": 8,
                "fix_img_size": False,
                "return_img_data": False,
                "include_tensors": False,
                "path_output": None,
                "tensor": None,
                "include_predictors": None,
                "exclude_predictors": None,
                "skip_detector": False,
            }
            for name, value in zip(positional_names, legacy_args):
                current = legacy_values[name]
                default = default_values[name]
                is_default = current is None if default is None else current == default
                if not is_default:
                    raise TypeError(
                        f"run_legacy() got multiple values for argument {name!r}"
                    )
                legacy_values[name] = value

            path_image = legacy_values["path_image"]
            batch_size = legacy_values["batch_size"]
            fix_img_size = legacy_values["fix_img_size"]
            return_img_data = legacy_values["return_img_data"]
            include_tensors = legacy_values["include_tensors"]
            path_output = legacy_values["path_output"]
            tensor = legacy_values["tensor"]
            include_predictors = legacy_values["include_predictors"]
            exclude_predictors = legacy_values["exclude_predictors"]
            skip_detector = legacy_values["skip_detector"]

        # v0.5 and v0.6 selected the first populated source in this order:
        # image_source, path_image, tensor. Normalize every alias combination
        # before the stricter v1 input boundary.
        if image_source is not None:
            path_image = None
            tensor = None
        elif path_image is not None:
            tensor = None

        result = self.run(
            image_source=image_source,
            path_image=path_image,
            face_batch_size=batch_size,
            fix_img_size=fix_img_size,
            include_tensors=include_tensors,
            path_output=path_output,
            tensor=tensor,
            include_predictors=include_predictors,
            exclude_predictors=exclude_predictors,
            skip_detector=skip_detector,
            input_policy=input_policy,
            input_spec=input_spec,
        )
        if not return_img_data:
            return Response(faces=result.faces, version=result.version)

        return ImageData(
            path_input=result.path_input,
            path_output=result.path_output,
            img=result.image if result.image is not None else torch.tensor([]),
            tensor=result.tensor if result.tensor is not None else torch.tensor([]),
            dims=result.dimensions,
            det=result.detection if result.detection is not None else Detection(),
            faces=result.faces,
            version=result.version,
            warnings=list(result.warnings),
        )

    def _read_input(
        self,
        image_source: Union[
            str, os.PathLike, torch.Tensor, np.ndarray, bytes, Image.Image
        ],
        fix_img_size: bool,
        *,
        input_policy: str = "coerce",
        input_spec: Optional[InputSpec] = None,
    ) -> ImageData:
        """Delegate every source type to the configured public reader entry point."""
        run = self.reader.run
        if self.__dict__.get("_reader_signature_owner") is self.reader:
            parameters = self.__dict__.get("_reader_signature_parameters")
        else:
            try:
                parameters = inspect.signature(run).parameters
            except (TypeError, ValueError) as exc:
                raise ConfigurationError(
                    "Configured reader.run must expose an inspectable public signature."
                ) from exc
            self._reader_signature_owner = self.reader
            self._reader_signature_parameters = parameters

        accepts_kwargs = any(
            parameter.kind == inspect.Parameter.VAR_KEYWORD
            for parameter in parameters.values()
        )
        reader_kwargs = {}
        if "fix_img_size" in parameters or accepts_kwargs:
            reader_kwargs["fix_img_size"] = fix_img_size

        supports_policy = "input_policy" in parameters or accepts_kwargs
        supports_spec = "input_spec" in parameters or accepts_kwargs
        if not supports_policy and input_policy != "coerce":
            raise ConfigurationError(
                "Configured reader uses the legacy protocol and cannot honor strict mode."
            )
        if not supports_spec and input_spec is not None:
            raise ConfigurationError(
                "Configured reader uses the legacy protocol and cannot honor InputSpec."
            )
        if supports_policy:
            reader_kwargs["input_policy"] = input_policy
        if supports_spec:
            reader_kwargs["input_spec"] = input_spec

        data = run(image_source, **reader_kwargs)
        if not isinstance(data, ImageData):
            raise ConfigurationError(
                "Configured reader.run must return facetorch.datastruct.ImageData."
            )
        if not supports_policy or not supports_spec:
            message = (
                "Configured reader uses the deprecated v0.x protocol; add keyword-only "
                "input_policy and input_spec parameters."
            )
            warnings.warn(message, DeprecationWarning, stacklevel=3)
            data.warnings.append(message)

        self._validate_reader_output(data)
        return data

    @staticmethod
    def _validate_reader_output(data: ImageData) -> None:
        tensor = data.tensor
        if not isinstance(tensor, torch.Tensor) or tensor.ndim != 4:
            raise ConfigurationError("Reader output tensor must have BCHW rank 4.")
        if tensor.shape[0] != 1:
            raise InputError(
                "Batched image input is not supported. Expected B=1, "
                f"got B={tensor.shape[0]}."
            )
        if tensor.shape[1] != 3:
            raise ConfigurationError(
                "Reader output must use the canonical three-channel RGB representation."
            )
        if tensor.dtype != torch.float32:
            raise ConfigurationError("Reader output tensor must use float32 values.")
        if not getattr(data, "_facetorch_canonical", False):
            if not torch.isfinite(tensor).all():
                raise InputError("Reader output contains NaN or Inf values.")
            if tensor.numel() and (
                float(tensor.min()) < 0.0 or float(tensor.max()) > 255.0
            ):
                raise ConfigurationError("Reader output values must stay within 0..255.")

FaceAnalyzer is the main class that reads images, runs face detection, tensor unification and facial feature prediction. It also draws bounding boxes and facial landmarks over the image.

The following components are used:

1. Reader - reads the image and returns an ImageData object containing the image tensor.
2. Detector - wrapper around a neural network that detects faces.
3. Unifier - processor that unifies sizes of all faces and normalizes them between 0 and 1.
4. Predictor dict - dict of wrappers around neural networks trained to analyze facial features.
5. Utilizer dict - dict of utilizer processors that can for example extract 3D face landmarks or draw boxes over the image.

Args

-----=

cfg : OmegaConf

Config object with image reader, face detector, unifier and predictor configurations.

Attributes

-----=

cfg : OmegaConf

Config object with image reader, face detector, unifier and predictor configurations.

reader : BaseReader

Reader object that reads the image and returns an ImageData object containing the image tensor.

detector : FaceDetector

Lazily loaded and cached FaceDetector object.

unifier : FaceUnifier

FaceUnifier object that unifies sizes of all faces and normalizes them between 0 and 1.

predictors : MutableMapping[str, FacePredictor]

Mapping of lazily loaded and cached predictors. Iterating names does not load models; accessing a value does.

utilizers : MutableMapping[str, FaceUtilizer]

Mapping of lazily loaded utilizer objects. Selection-linked utilizers load only when their predictor ran.

logger : logging.Logger

Logger object that logs messages to the console or to a file.

Instance variables

prop detector

Expand source code

@property
def detector(self):
    """Return the configured detector, constructing and caching it on demand."""
    detector = self.__dict__.get("_detector", _UNLOADED)
    if detector is not _UNLOADED:
        return detector

    detector_config = self.__dict__.get("_detector_config")
    if detector_config is None:
        raise ConfigurationError("No face detector is configured.")

    lock = self._get_component_lock()
    with lock:
        if self._detector is _UNLOADED:
            self.logger.info("Initializing FaceDetector")
            self._detector = instantiate(detector_config)
    return self._detector

Return the configured detector, constructing and caching it on demand.

prop predictors : MutableMapping[str, FacePredictor]

Expand source code

@property
def predictors(self) -> MutableMapping[str, FacePredictor]:
    """Return the lazy predictor mapping without constructing its values."""
    registry = self.__dict__.get("_predictors")
    if registry is None:
        registry = _LazyComponentRegistry(lock=self._get_component_lock())
        self._predictors = registry
    return registry

Return the lazy predictor mapping without constructing its values.

prop configured_predictors : tuple[str, ...]

Expand source code

@property
def configured_predictors(self) -> tuple[str, ...]:
    """Predictor names in deterministic configuration order, without loading."""
    return tuple(self.predictors)

Predictor names in deterministic configuration order, without loading.

prop loaded_predictors : tuple[str, ...]

Expand source code

@property
def loaded_predictors(self) -> tuple[str, ...]:
    """Predictor names whose wrappers and models are already cached."""
    registry = self.predictors
    if isinstance(registry, _LazyComponentRegistry):
        return registry.loaded_names
    return tuple(registry)

Predictor names whose wrappers and models are already cached.

prop utilizers : MutableMapping[str, typing.Any]

Expand source code

@property
def utilizers(self) -> MutableMapping[str, Any]:
    """Return the lazy utilizer mapping without constructing its values."""
    registry = self.__dict__.get("_utilizers")
    if registry is None:
        registry = _LazyComponentRegistry(lock=self._get_component_lock())
        self._utilizers = registry
    return registry

Return the lazy utilizer mapping without constructing its values.

prop configured_utilizers : tuple[str, ...]

Expand source code

@property
def configured_utilizers(self) -> tuple[str, ...]:
    """Utilizer names in deterministic configuration order, without loading."""
    return tuple(self.utilizers)

Utilizer names in deterministic configuration order, without loading.

prop loaded_utilizers : tuple[str, ...]

Expand source code

@property
def loaded_utilizers(self) -> tuple[str, ...]:
    """Utilizer names whose objects are already cached."""
    registry = self.utilizers
    if isinstance(registry, _LazyComponentRegistry):
        return registry.loaded_names
    return tuple(registry)

Utilizer names whose objects are already cached.

prop utilizer_dependencies : dict[str, tuple[str, ...]]

Expand source code

@property
def utilizer_dependencies(self) -> dict[str, tuple[str, ...]]:
    """Explicit predictor requirements for configured utilizers."""
    return dict(self.__dict__.get("_utilizer_dependencies", {}))

Explicit predictor requirements for configured utilizers.

prop detector_loaded : bool

Expand source code

@property
def detector_loaded(self) -> bool:
    """Whether the detector wrapper and model are already cached."""
    return self.__dict__.get("_detector", _UNLOADED) is not _UNLOADED

Whether the detector wrapper and model are already cached.

Methods

def run(self,
image_source: str | os.PathLike | torch.Tensor | numpy.ndarray | bytes | PIL.Image.Image | None = None,
path_image: str | None = None,
face_batch_size: int | None = None,
fix_img_size: bool = False,
return_img_data: bool | None = None,
include_tensors: bool = False,
path_output: str | None = None,
tensor: torch.Tensor | None = None,
include_predictors: List[str] | None = None,
exclude_predictors: List[str] | None = None,
skip_detector: bool = False,
*,
batch_size: int | None = None,
input_policy: str = 'coerce',
input_spec: InputSpec | None = None) ‑> AnalysisResult

Expand source code

@Timer("FaceAnalyzer.run", "{name}: {milliseconds:.2f} ms", logger=logger.debug)
def run(
    self,
    image_source: Optional[
        Union[str, os.PathLike, torch.Tensor, np.ndarray, bytes, Image.Image]
    ] = None,
    path_image: Optional[str] = None,
    face_batch_size: Optional[int] = None,
    fix_img_size: bool = False,
    return_img_data: Optional[bool] = None,
    include_tensors: bool = False,
    path_output: Optional[str] = None,
    tensor: Optional[torch.Tensor] = None,
    include_predictors: Optional[List[str]] = None,
    exclude_predictors: Optional[List[str]] = None,
    skip_detector: bool = False,
    *,
    batch_size: Optional[int] = None,
    input_policy: str = "coerce",
    input_spec: Optional[InputSpec] = None,
) -> AnalysisResult:
    """Analyze exactly one source image and return one stable result type.

    Args:
        image_source: Input accepted by the configured reader. The default
            reader accepts local paths, tensors, NumPy arrays, bytes, and PIL
            images. URLs require an explicit URLReader configuration.
        path_image (Optional[str]): Deprecated. Use image_source instead.
        face_batch_size (Optional[int]): Number of faces from this image sent to
            each predictor at once. This is an upper bound; predictors may use
            smaller chunks to honor their model artifact. Default: 8.
        fix_img_size (bool): If True, resizes the image to the size specified in reader. Default is False.
        return_img_data (Optional[bool]): Deprecated no-op. Use
            ``include_tensors`` and the fields on ``AnalysisResult`` or call
            ``run_legacy`` for the former flag-dependent return type.
        include_tensors (bool): If True, includes tensors in the returned data object. If False, tensors are removed. Default is False.
        path_output (Optional[str]): Path where to save the image with detected faces. If None, the image is not saved. Default: None.
        tensor (Optional[torch.Tensor]): Deprecated. Use image_source instead.
        include_predictors (Optional[List[str]]): Names to run. None runs all
            configured predictors and an empty collection runs none.
        exclude_predictors (Optional[List[str]]): Names to omit. None and an
            empty collection omit none. Cannot be combined with an include.
        skip_detector (bool): If True, skip face detection, avoid constructing
            its model, and treat the input as a pre-cropped face. Default: False.
        batch_size (Optional[int]): Deprecated warning alias for
            ``face_batch_size`` throughout v1.x.
        input_policy (str): ``coerce`` (default) or ``strict``.
        input_spec (Optional[InputSpec]): Explicit source layout/range/color
            description, especially for strict-mode conversions.

    Returns:
        AnalysisResult: Stable result for the one source image.

    """

    compatibility_warnings = []

    if face_batch_size is not None and batch_size is not None:
        raise ConfigurationError(
            "Specify only face_batch_size; batch_size is its deprecated alias."
        )
    if batch_size is not None:
        message = (
            "batch_size is deprecated and will be removed after v1.x; "
            "use face_batch_size."
        )
        warnings.warn(message, DeprecationWarning, stacklevel=2)
        compatibility_warnings.append(message)
        effective_face_batch_size = batch_size
    elif face_batch_size is None:
        effective_face_batch_size = 8
    else:
        effective_face_batch_size = face_batch_size

    if (
        isinstance(effective_face_batch_size, bool)
        or not isinstance(effective_face_batch_size, int)
        or effective_face_batch_size < 1
    ):
        raise ConfigurationError(
            "face_batch_size must be an integer greater than or equal to 1, "
            f"got {effective_face_batch_size!r}."
        )

    if return_img_data is not None:
        message = (
            "return_img_data no longer changes FaceAnalyzer.run's return type; "
            "use include_tensors or the explicit run_legacy adapter."
        )
        warnings.warn(message, DeprecationWarning, stacklevel=2)
        compatibility_warnings.append(message)

    def _run_component(label, operation):
        try:
            return operation()
        except FacetorchError:
            raise
        except Exception as exc:
            raise InferenceError(f"{label} failed during analysis.") from exc

    def _predict_batch(
        data: ImageData, predictor: FacePredictor, predictor_name: str
    ) -> ImageData:
        n_faces = len(data.faces)
        predictor_limit = getattr(predictor, "max_batch_size", None)
        if predictor_limit is not None and (
            isinstance(predictor_limit, bool)
            or not isinstance(predictor_limit, int)
            or predictor_limit < 1
        ):
            raise ConfigurationError(
                f"Face predictor {predictor_name!r} has invalid "
                f"max_batch_size {predictor_limit!r}."
            )
        chunk_size = (
            effective_face_batch_size
            if predictor_limit is None
            else min(effective_face_batch_size, predictor_limit)
        )

        for face_indx_start in range(0, n_faces, chunk_size):
            face_indx_end = min(face_indx_start + chunk_size, n_faces)

            face_batch_tensor = torch.stack(
                [face.tensor for face in data.faces[face_indx_start:face_indx_end]]
            )
            preds = predictor.run(face_batch_tensor)
            expected_count = face_indx_end - face_indx_start
            try:
                prediction_count = len(preds)
            except TypeError as exc:
                raise InferenceError(
                    f"Face predictor {predictor_name!r} must return one "
                    "prediction per input face; its result has no length."
                ) from exc
            if prediction_count != expected_count:
                raise InferenceError(
                    f"Face predictor {predictor_name!r} returned "
                    f"{prediction_count} prediction(s) for "
                    f"{expected_count} input face(s)."
                )
            data.add_preds(preds, predictor_name, face_indx_start)

        return data

    self.logger.info("Running FaceAnalyzer")
    selected_predictors = self._select_predictor_names(
        include_predictors, exclude_predictors
    )
    configured_predictors = set(self.configured_predictors)
    if skip_detector and selected_predictors and self.unifier is None:
        raise ConfigurationError(
            "skip_detector=True with selected predictors requires a face "
            "unifier. Configure analyzer.unifier or pass "
            "include_predictors=[] for predictor-free processing."
        )

    supplied_sources = [
        name
        for name, value in (
            ("image_source", image_source),
            ("path_image", path_image),
            ("tensor", tensor),
        )
        if value is not None
    ]
    if len(supplied_sources) > 1:
        raise InputError(
            "Supply exactly one input source using image_source. Received: "
            + ", ".join(supplied_sources)
            + "."
        )
    if not supplied_sources:
        raise InputError(
            "image_source is required. Pass a file path, URL, tensor, numpy array, "
            "bytes, or PIL Image."
        )
    if path_image is not None:
        message = "path_image is deprecated; use image_source."
        warnings.warn(message, DeprecationWarning, stacklevel=2)
        compatibility_warnings.append(message)
        image_source = path_image
    elif tensor is not None:
        message = "tensor is deprecated; use image_source."
        warnings.warn(message, DeprecationWarning, stacklevel=2)
        compatibility_warnings.append(message)
        image_source = tensor

    self.logger.info("Reading image")
    data = self._read_input(
        image_source,
        fix_img_size,
        input_policy=input_policy,
        input_spec=input_spec,
    )
    data.warnings.extend(compatibility_warnings)

    path_output = None if path_output == "None" else path_output
    data.path_output = path_output

    try:
        data.version = version("facetorch")
    except Exception as e:
        self.logger.warning("Could not get version number", extra={"error": e})

    if skip_detector:
        self.logger.info("Skipping detector (skip_detector=True)")
        face_tensor = data.tensor[0]
        face = Face(
            indx=0,
            loc=Location(
                x1=0, y1=0, x2=data.dims.width, y2=data.dims.height
            ),
            dims=Dimensions(
                height=data.dims.height, width=data.dims.width
            ),
            tensor=face_tensor,
            ratio=1.0,
        )
        data.faces = [face]
        n_faces = 1
    else:
        self.logger.info("Detecting faces")
        data = _run_component("Face detector", lambda: self.detector.run(data))
        n_faces = len(data.faces)

    self.logger.info(f"Number of faces: {n_faces}")

    ran_predictors = set()
    if n_faces > 0:
        if selected_predictors and self.unifier is None:
            raise ConfigurationError(
                "Detected faces cannot be sent to selected predictors without "
                "a face unifier. Configure analyzer.unifier or pass "
                "include_predictors=[] for predictor-free processing."
            )

        if self.unifier is not None:
            self.logger.info("Unifying faces")
            data = _run_component("Face unifier", lambda: self.unifier.run(data))

        self.logger.info("Predicting facial features")
        for predictor_name in selected_predictors:
            self.logger.info(f"Running FacePredictor: {predictor_name}")
            data = _run_component(
                f"Face predictor {predictor_name!r}",
                lambda predictor_name=predictor_name: _predict_batch(
                    data,
                    self.predictors[predictor_name],
                    predictor_name,
                ),
            )
            ran_predictors.add(predictor_name)

        utilizer_names = tuple(self.utilizers)
    else:
        utilizer_names = ("save",) if "save" in self.utilizers else ()

    self.logger.info("Utilizing facial features")
    dependencies = self.__dict__.get("_utilizer_dependencies", {})
    for utilizer_name in utilizer_names:
        required_predictors = set(dependencies.get(utilizer_name, ()))
        unknown_requirements = required_predictors - configured_predictors
        if unknown_requirements:
            raise ConfigurationError(
                f"Utilizer {utilizer_name!r} requires unknown predictor(s): "
                + ", ".join(sorted(unknown_requirements))
                + "."
            )
        missing_predictors = required_predictors - ran_predictors
        if missing_predictors:
            self.logger.info(
                f"Skipping BaseUtilizer: {utilizer_name} "
                "(required predictor(s) not run: "
                + ", ".join(sorted(missing_predictors))
                + ")"
            )
            continue
        self.logger.info(f"Running BaseUtilizer: {utilizer_name}")
        data = _run_component(
            f"Face utilizer {utilizer_name!r}",
            lambda utilizer_name=utilizer_name: self.utilizers[
                utilizer_name
            ].run(data),
        )

    if not include_tensors:
        self.logger.debug(
            "Removing tensors from response as include_tensors is False"
        )
        data.reset_tensors()

    result = AnalysisResult.from_image_data(data, include_tensors=include_tensors)
    self.logger.debug(
        "Returning analysis result",
        extra={"face_count": len(result.faces), "version": result.version},
    )
    return result

Analyze exactly one source image and return one stable result type.

Args

-----=

image_source

Input accepted by the configured reader. The default reader accepts local paths, tensors, NumPy arrays, bytes, and PIL images. URLs require an explicit URLReader configuration.

path_image : Optional[str]

Deprecated. Use image_source instead.

face_batch_size : Optional[int]

Number of faces from this image sent to each predictor at once. This is an upper bound; predictors may use smaller chunks to honor their model artifact. Default: 8.

fix_img_size : bool

If True, resizes the image to the size specified in reader. Default is False.

return_img_data : Optional[bool]

Deprecated no-op. Use include_tensors and the fields on AnalysisResult or call run_legacy for the former flag-dependent return type.

include_tensors : bool

If True, includes tensors in the returned data object. If False, tensors are removed. Default is False.

path_output : Optional[str]

Path where to save the image with detected faces. If None, the image is not saved. Default: None.

tensor : Optional[torch.Tensor]

Deprecated. Use image_source instead.

include_predictors : Optional[List[str]]

Names to run. None runs all configured predictors and an empty collection runs none.

exclude_predictors : Optional[List[str]]

Names to omit. None and an empty collection omit none. Cannot be combined with an include.

skip_detector : bool

If True, skip face detection, avoid constructing its model, and treat the input as a pre-cropped face. Default: False.

batch_size : Optional[int]

Deprecated warning alias for face_batch_size throughout v1.x.

input_policy : str

coerce (default) or strict.

input_spec : Optional[InputSpec]

Explicit source layout/range/color description, especially for strict-mode conversions.

Returns

-----=

AnalysisResult

Stable result for the one source image.

def run_legacy(self,
image_source: str | os.PathLike | torch.Tensor | numpy.ndarray | bytes | PIL.Image.Image | None = None,
*legacy_args: Any,
path_image: str | None = None,
batch_size: int = 8,
fix_img_size: bool = False,
return_img_data: bool = False,
include_tensors: bool = False,
path_output: str | None = None,
tensor: torch.Tensor | None = None,
include_predictors: List[str] | None = None,
exclude_predictors: List[str] | None = None,
skip_detector: bool = False,
input_policy: str = 'coerce',
input_spec: InputSpec | None = None) ‑> Response | ImageData

Expand source code

def run_legacy(
    self,
    image_source: Optional[
        Union[str, os.PathLike, torch.Tensor, np.ndarray, bytes, Image.Image]
    ] = None,
    *legacy_args: Any,
    path_image: Optional[str] = None,
    batch_size: int = 8,
    fix_img_size: bool = False,
    return_img_data: bool = False,
    include_tensors: bool = False,
    path_output: Optional[str] = None,
    tensor: Optional[torch.Tensor] = None,
    include_predictors: Optional[List[str]] = None,
    exclude_predictors: Optional[List[str]] = None,
    skip_detector: bool = False,
    input_policy: str = "coerce",
    input_spec: Optional[InputSpec] = None,
) -> Union[Response, ImageData]:
    """Run the canonical pipeline using either shipped v0 positional order.

    Releases through v0.4 used ``path_image, batch_size, ...``. Releases
    v0.5 and v0.6 inserted ``image_source, path_image, batch_size, ...``.
    This adapter accepts both layouts and preserves their flag-dependent
    return type. Prefer ``run()`` for new integrations.
    """
    warnings.warn(
        "FaceAnalyzer.run_legacy is a v1.x compatibility adapter; migrate to run().",
        DeprecationWarning,
        stacklevel=2,
    )

    if legacy_args:
        # The second positional value distinguishes the two historical
        # layouts: v0.5+ path_image was a path/None, while earlier releases
        # placed the integer batch_size there.
        has_v05_source_slot = legacy_args[0] is None or isinstance(
            legacy_args[0], (str, os.PathLike)
        )
        positional_names = (
            (
                "path_image",
                "batch_size",
                "fix_img_size",
                "return_img_data",
                "include_tensors",
                "path_output",
                "tensor",
                "include_predictors",
                "exclude_predictors",
                "skip_detector",
            )
            if has_v05_source_slot
            else (
                "batch_size",
                "fix_img_size",
                "return_img_data",
                "include_tensors",
                "path_output",
                "tensor",
                "include_predictors",
                "exclude_predictors",
                "skip_detector",
            )
        )
        if len(legacy_args) > len(positional_names):
            maximum = len(positional_names) + 1
            supplied = len(legacy_args) + 1
            raise TypeError(
                "run_legacy() takes at most "
                f"{maximum} positional arguments but {supplied} were given"
            )

        legacy_values = {
            "path_image": path_image,
            "batch_size": batch_size,
            "fix_img_size": fix_img_size,
            "return_img_data": return_img_data,
            "include_tensors": include_tensors,
            "path_output": path_output,
            "tensor": tensor,
            "include_predictors": include_predictors,
            "exclude_predictors": exclude_predictors,
            "skip_detector": skip_detector,
        }
        default_values = {
            "path_image": None,
            "batch_size": 8,
            "fix_img_size": False,
            "return_img_data": False,
            "include_tensors": False,
            "path_output": None,
            "tensor": None,
            "include_predictors": None,
            "exclude_predictors": None,
            "skip_detector": False,
        }
        for name, value in zip(positional_names, legacy_args):
            current = legacy_values[name]
            default = default_values[name]
            is_default = current is None if default is None else current == default
            if not is_default:
                raise TypeError(
                    f"run_legacy() got multiple values for argument {name!r}"
                )
            legacy_values[name] = value

        path_image = legacy_values["path_image"]
        batch_size = legacy_values["batch_size"]
        fix_img_size = legacy_values["fix_img_size"]
        return_img_data = legacy_values["return_img_data"]
        include_tensors = legacy_values["include_tensors"]
        path_output = legacy_values["path_output"]
        tensor = legacy_values["tensor"]
        include_predictors = legacy_values["include_predictors"]
        exclude_predictors = legacy_values["exclude_predictors"]
        skip_detector = legacy_values["skip_detector"]

    # v0.5 and v0.6 selected the first populated source in this order:
    # image_source, path_image, tensor. Normalize every alias combination
    # before the stricter v1 input boundary.
    if image_source is not None:
        path_image = None
        tensor = None
    elif path_image is not None:
        tensor = None

    result = self.run(
        image_source=image_source,
        path_image=path_image,
        face_batch_size=batch_size,
        fix_img_size=fix_img_size,
        include_tensors=include_tensors,
        path_output=path_output,
        tensor=tensor,
        include_predictors=include_predictors,
        exclude_predictors=exclude_predictors,
        skip_detector=skip_detector,
        input_policy=input_policy,
        input_spec=input_spec,
    )
    if not return_img_data:
        return Response(faces=result.faces, version=result.version)

    return ImageData(
        path_input=result.path_input,
        path_output=result.path_output,
        img=result.image if result.image is not None else torch.tensor([]),
        tensor=result.tensor if result.tensor is not None else torch.tensor([]),
        dims=result.dimensions,
        det=result.detection if result.detection is not None else Detection(),
        faces=result.faces,
        version=result.version,
        warnings=list(result.warnings),
    )

Run the canonical pipeline using either shipped v0 positional order.

Releases through v0.4 used path_image, batch_size, …. Releases v0.5 and v0.6 inserted image_source, path_image, batch_size, …. This adapter accepts both layouts and preserves their flag-dependent return type. Prefer run() for new integrations.