Package facetorch

def load_config(
    profile: Optional[str] = "cpu",
    *,
    overrides: Optional[Sequence[str]] = None,
    offline: Optional[bool] = None,
    allow_legacy_models: Optional[bool] = None,
) -> DictConfig:
    """Compose packaged facetorch defaults independently of the current directory.

    Args:
        profile: ``"cpu"`` (default), ``"gpu"``, or ``None`` to retain the
            packaged device value.
        overrides: Hydra override strings applied after the selected profile.
        offline: Explicitly disable or allow model network access. ``None`` uses
            the packaged environment-backed default.
        allow_legacy_models: Explicit opt-in for eligible verified TorchScript
            fallback artifacts. ``None`` retains the packaged false default.

    Returns:
        A fully composed configuration with ``cfg.analyzer`` available.
    """
    register_path_resolvers()
    option_overrides = _option_overrides(
        _normalize_overrides(overrides),
        offline=offline,
        allow_legacy_models=allow_legacy_models,
    )
    composed_overrides = _profile_overrides(profile, option_overrides)
    try:
        with initialize_config_module(
            config_module="facetorch.configs",
            version_base=None,
            job_name="facetorch-load-config",
        ):
            return compose(config_name="config", overrides=composed_overrides)
    except ConfigurationError:
        raise
    except Exception as exc:
        raise ConfigurationError(
            "Could not compose packaged facetorch configuration."
        ) from exc

def load_config_from_path(
    path: ConfigPath,
    *,
    profile: Optional[str] = None,
    overrides: Optional[Sequence[str]] = None,
    offline: Optional[bool] = None,
    allow_legacy_models: Optional[bool] = None,
) -> DictConfig:
    """Compose an advanced external Hydra YAML tree from an explicit file path.

    Relative paths are resolved against the caller's current directory. The file's
    parent becomes Hydra's configuration directory, so its ``defaults`` list and
    sibling configuration groups are composed normally.
    """
    register_path_resolvers()
    config_file = Path(path).expanduser().resolve()
    if config_file.suffix.lower() not in {".yaml", ".yml"}:
        raise ConfigurationError(
            "External configuration must be a .yaml or .yml file."
        )
    if not config_file.is_file():
        raise ConfigurationError(
            f"External configuration file does not exist: {config_file}."
        )

    option_overrides = _option_overrides(
        _normalize_overrides(overrides),
        offline=offline,
        allow_legacy_models=allow_legacy_models,
    )
    composed_overrides = _profile_overrides(profile, option_overrides)
    try:
        with initialize_config_dir(
            config_dir=os.fspath(config_file.parent),
            version_base=None,
            job_name="facetorch-load-external-config",
        ):
            return compose(
                config_name=config_file.name[: -len(config_file.suffix)],
                overrides=composed_overrides,
            )
    except ConfigurationError:
        raise
    except Exception as exc:
        raise ConfigurationError(
            f"Could not compose external configuration: {config_file}."
        ) from exc

def get_cache_dir() -> Path:
    """Return the configured OS-appropriate facetorch cache root without creating it."""
    configured = os.environ.get(CACHE_DIR_ENV)
    if configured:
        return _normalized_path(configured)
    return _default_cache_dir()

def get_metadata_dir() -> Path:
    """Return the versioned generated/model metadata location without creating it."""
    configured = os.environ.get(METADATA_DIR_ENV)
    if configured:
        return _normalized_path(configured)
    return get_cache_dir() / "metadata" / "v1"

def get_model_dir() -> Path:
    """Return the versioned model cache location without creating it."""
    configured = os.environ.get(MODEL_DIR_ENV)
    if configured:
        return _normalized_path(configured)
    return get_cache_dir() / "models" / "v1"

def get_offline_mode(*, environ: Optional[Mapping[str, str]] = None) -> bool:
    """Return whether model network access is disabled by the public environment flag."""
    env = os.environ if environ is None else environ
    raw = str(env.get(OFFLINE_ENV, "")).strip().lower()
    if raw in _TRUE_VALUES:
        return True
    if raw in _FALSE_VALUES:
        return False
    raise ConfigurationError(
        f"{OFFLINE_ENV} must be one of 1/0, true/false, yes/no, or on/off; "
        f"got {env.get(OFFLINE_ENV)!r}."
    )

def cleanup_quarantined_cache(
    root: Optional[str | os.PathLike] = None,
    *,
    confirm: bool = False,
) -> CacheCleanupReport:
    """Delete only reported quarantine files and only after explicit confirmation."""
    report = inspect_quarantined_cache(root)
    if not confirm:
        return report
    for path in report.paths:
        path.unlink()
    return CacheCleanupReport(
        paths=report.paths,
        total_bytes=report.total_bytes,
        deleted=True,
    )

def inspect_incompatible_cache(
    root: Optional[str | os.PathLike] = None,
) -> CacheCleanupReport:
    """Report persisted runtime/schema rejections without changing the cache."""
    model_root = get_model_dir().resolve()
    selected = model_root if root is None else Path(root).expanduser().resolve()
    if selected != model_root and not selected.is_relative_to(model_root):
        raise ConfigurationError(
            "Incompatibility reset is restricted to facetorch's versioned model "
            "cache directory."
        )
    paths = (
        tuple(sorted(selected.rglob(".incompatible.json")))
        if selected.exists()
        else ()
    )
    files = tuple(path for path in paths if path.is_file())
    return CacheCleanupReport(
        paths=files,
        total_bytes=sum(path.stat().st_size for path in files),
        deleted=False,
    )

def inspect_legacy_cache(path: str | os.PathLike) -> tuple[CacheEntryInspection, ...]:
    """Hash and classify old model files without deserializing or executing them."""
    root = Path(path).expanduser()
    if not root.exists():
        raise ConfigurationError(f"Legacy cache path does not exist: {root}.")
    candidates = [root] if root.is_file() else sorted(root.rglob("*"))
    entries = []
    for candidate in candidates:
        if not candidate.is_file() or candidate.suffix.lower() not in {".pt", ".pt2"}:
            continue
        detected = detect_model_format(candidate)
        entries.append(
            CacheEntryInspection(
                path=candidate,
                size_bytes=candidate.stat().st_size,
                sha256=sha256_file(candidate),
                detected_format=detected,
                mislabeled=candidate.suffix.lower() == ".pt2"
                and detected == "torchscript",
            )
        )
    return tuple(entries)

def inspect_quarantined_cache(
    root: Optional[str | os.PathLike] = None,
) -> CacheCleanupReport:
    """Report quarantined entries and reclaimable bytes without deleting anything."""
    paths = []
    for cache_root in _allowed_quarantine_roots(root):
        if cache_root.exists():
            paths.extend(
                path
                for path in cache_root.rglob("*.quarantine.*")
                if path.is_file()
            )
    unique_paths = tuple(sorted(set(paths)))
    return CacheCleanupReport(
        paths=unique_paths,
        total_bytes=sum(path.stat().st_size for path in unique_paths),
        deleted=False,
    )

def migrate_legacy_artifact(
    source: str | os.PathLike,
    artifact_id: str,
    destination: str | os.PathLike,
) -> Path:
    """Copy one exact manifest match into v1 layout without changing the source."""
    source_path = Path(source).expanduser()
    destination_path = Path(destination).expanduser()
    descriptor = get_model_manifest().descriptor(artifact_id)
    if destination_path.name != descriptor.filename:
        raise ConfigurationError(
            f"Migration destination must preserve the authenticated filename "
            f"{descriptor.filename!r}."
        )
    verify_artifact(source_path, descriptor)
    if destination_path.exists():
        try:
            return verify_artifact(destination_path, descriptor)
        except ArtifactIntegrityError as exc:
            raise ArtifactIntegrityError(
                f"Migration destination already exists and is not the requested "
                f"artifact: {destination_path}."
            ) from exc
    destination_path.parent.mkdir(parents=True, exist_ok=True)
    temporary_path: Optional[Path] = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix=f".{destination_path.name}.",
            suffix=".tmp",
            dir=destination_path.parent,
            delete=False,
        ) as temporary:
            temporary_path = Path(temporary.name)
            with source_path.open("rb") as source_file:
                shutil.copyfileobj(source_file, temporary, length=1024 * 1024)
            temporary.flush()
            os.fsync(temporary.fileno())
        verify_artifact(temporary_path, descriptor)
        os.replace(temporary_path, destination_path)
    finally:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)
    return verify_artifact(destination_path, descriptor)

def plan_model_prefetch(
    profile: str = "cpu",
    *,
    include_predictors: Optional[Iterable[str]] = None,
    skip_detector: bool = False,
    offline: Optional[bool] = None,
    allow_legacy_models: bool = False,
    overrides: Optional[Sequence[str]] = None,
) -> PrefetchPlan:
    """Resolve exact artifacts and costs without creating files or using the network."""
    if not isinstance(skip_detector, bool):
        raise ConfigurationError("skip_detector must be a boolean.")
    cfg = load_config(
        profile,
        overrides=overrides,
        offline=offline,
        allow_legacy_models=allow_legacy_models,
    )
    predictor_names = _selected_predictors(cfg, include_predictors)
    selected_configs: list[tuple[str, object]] = []
    if not skip_detector and "detector" in cfg.analyzer:
        selected_configs.append(("detector", cfg.analyzer.detector.downloader))
    selected_configs.extend(
        (f"predictor.{name}", cfg.analyzer.predictor[name].downloader)
        for name in predictor_names
    )

    manifest = get_model_manifest()
    items: list[PrefetchItem] = []
    for component, downloader in selected_configs:
        sidecar = (
            Path(str(downloader.path_local)).expanduser().parent
            / ".incompatible.json"
        )
        key = incompatibility_key(
            manifest.manifest_revision,
            str(torch.__version__),
            str(downloader.device),
        )
        try:
            incompatible = read_incompatible_artifact_ids(sidecar, key)
        except ArtifactIntegrityError:
            # Planning is deliberately non-mutating. Runtime resolution will
            # quarantine the malformed sidecar and make this same empty choice.
            incompatible = set()
        candidates = manifest.candidates(
            str(downloader.manifest_id),
            torch_version=str(torch.__version__),
            device=str(downloader.device),
            allow_legacy_models=allow_legacy_models,
            incompatible_artifact_ids=incompatible,
        )
        descriptor = candidates[0]
        path = descriptor.cache_path(str(downloader.path_local))
        items.append(
            PrefetchItem(
                component=component,
                artifact_id=descriptor.artifact_id,
                path=path,
                format=descriptor.format,
                size_bytes=descriptor.size_bytes,
                sha256=descriptor.sha256,
                cached=_is_verified(path, descriptor),
            )
        )
    if "align" in _selected_utilizers(cfg, predictor_names):
        items.append(_metadata_prefetch_item(cfg))
    return PrefetchPlan(profile=profile, items=tuple(items))

def prefetch_models(
    profile: str = "cpu",
    *,
    include_predictors: Optional[Iterable[str]] = None,
    skip_detector: bool = False,
    offline: Optional[bool] = None,
    allow_legacy_models: bool = False,
    overrides: Optional[Sequence[str]] = None,
    confirm: bool = False,
) -> PrefetchResult:
    """Download exactly a planned selection after explicit bulk-cost confirmation."""
    requested_predictors = (
        tuple(include_predictors)
        if include_predictors is not None
        and not isinstance(include_predictors, (str, bytes))
        else include_predictors
    )
    plan = plan_model_prefetch(
        profile,
        include_predictors=requested_predictors,
        skip_detector=skip_detector,
        offline=offline,
        allow_legacy_models=allow_legacy_models,
        overrides=overrides,
    )
    if plan.download_bytes and len(plan.items) > 1 and not confirm:
        mib = plan.download_bytes / (1024 * 1024)
        raise ConfigurationError(
            f"Prefetch would download approximately {mib:.1f} MiB across "
            f"{len(plan.items)} artifacts. Review plan_model_prefetch() and pass "
            "confirm=True to continue."
        )

    cfg = load_config(
        profile,
        overrides=overrides,
        offline=offline,
        allow_legacy_models=allow_legacy_models,
    )
    predictor_names = _selected_predictors(cfg, requested_predictors)
    downloader_configs = []
    if not skip_detector and "detector" in cfg.analyzer:
        downloader_configs.append(cfg.analyzer.detector.downloader)
    downloader_configs.extend(
        cfg.analyzer.predictor[name].downloader for name in predictor_names
    )
    if "align" in _selected_utilizers(cfg, predictor_names):
        downloader_configs.append(
            cfg.analyzer.utilizer.align.downloader_meta
        )

    paths = []
    for downloader_config in downloader_configs:
        downloader = instantiate(downloader_config)
        paths.append(Path(downloader.run()))
    return PrefetchResult(plan=plan, paths=tuple(paths))

def reset_incompatible_cache(
    root: Optional[str | os.PathLike] = None,
    *,
    confirm: bool = False,
) -> CacheCleanupReport:
    """Explicitly clear persisted runtime/schema rejections after remediation."""
    report = inspect_incompatible_cache(root)
    if not confirm:
        return report
    for path in report.paths:
        path.unlink()
    return CacheCleanupReport(
        paths=report.paths,
        total_bytes=report.total_bytes,
        deleted=True,
    )

@dataclass
class AnalysisResult:
    """Stable result for one analyzed source image.

    ``faces``, ``version``, dimensions, paths, and warnings are always available.
    Tensor-heavy fields are ``None`` unless ``include_tensors=True`` was used.
    Runtime timing remains diagnostic logging rather than a stable result field.
    ``img``, ``det``, and ``dims`` remain warning aliases throughout v1.x.
    """

    faces: List[Face] = field(default_factory=list)
    version: str = field(default_factory=str)
    image: Optional[torch.Tensor] = None
    tensor: Optional[torch.Tensor] = None
    detection: Optional[Detection] = None
    dimensions: Dimensions = field(default_factory=Dimensions)
    path_input: Optional[str] = None
    path_output: Optional[str] = None
    warnings: List[str] = field(default_factory=list)

    @classmethod
    def from_image_data(
        cls, data: ImageData, *, include_tensors: bool
    ) -> "AnalysisResult":
        """Create the public result without introducing a second pipeline."""
        return cls(
            faces=data.faces,
            version=data.version,
            image=data.img if include_tensors else None,
            tensor=data.tensor if include_tensors else None,
            detection=data.det if include_tensors else None,
            dimensions=data.dims,
            path_input=data.path_input,
            path_output=data.path_output,
            warnings=list(data.warnings),
        )

    @property
    def img(self) -> Optional[torch.Tensor]:
        """Deprecated v0.x alias for :attr:`image`."""
        _warnings.warn(
            "AnalysisResult.img is deprecated; use AnalysisResult.image.",
            DeprecationWarning,
            stacklevel=2,
        )
        return self.image

    @property
    def det(self) -> Optional[Detection]:
        """Deprecated v0.x alias for :attr:`detection`."""
        _warnings.warn(
            "AnalysisResult.det is deprecated; use AnalysisResult.detection.",
            DeprecationWarning,
            stacklevel=2,
        )
        return self.detection

    @property
    def dims(self) -> Dimensions:
        """Deprecated v0.x alias for :attr:`dimensions`."""
        _warnings.warn(
            "AnalysisResult.dims is deprecated; use AnalysisResult.dimensions.",
            DeprecationWarning,
            stacklevel=2,
        )
        return self.dimensions

class ArtifactIntegrityError(FacetorchError, RuntimeError):
    """A model or distribution artifact failed integrity verification."""

class CacheLockError(FacetorchError, TimeoutError):
    """A model-cache operation could not acquire its process lock."""

@dataclass(frozen=True)
class CacheCleanupReport:
    """Quarantine inventory and optional explicit cleanup result."""

    paths: tuple[Path, ...]
    total_bytes: int
    deleted: bool

@dataclass(frozen=True)
class CacheEntryInspection:
    """Non-executing inspection result for one possible legacy artifact."""

    path: Path
    size_bytes: int
    sha256: str
    detected_format: str
    mislabeled: bool

class ConfigurationError(FacetorchError, ValueError):
    """The configured component graph or option set is invalid."""

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.")

class FacetorchError(Exception):
    """Base class for actionable facetorch errors."""

@dataclass
class Dimensions:
    """Data class for image dimensions.

    Attributes:
        height (int): Image height.
        width (int): Image width.
    """

    height: int = field(default=0)
    width: int = field(default=0)

@dataclass
class Detection:
    """Data class for detector output.

    Attributes:
        loc (torch.Tensor): Locations of faces
        conf (torch.Tensor): Confidences of faces
        landmarks (torch.Tensor): Selected landmark coordinates in source-image space.
        boxes (torch.Tensor): Selected bounding boxes in source-image space.
        dets (torch.Tensor): Selected boxes and confidence scores.

    """

    loc: torch.Tensor = field(default_factory=torch.Tensor)
    conf: torch.Tensor = field(default_factory=torch.Tensor)
    landmarks: torch.Tensor = field(default_factory=torch.Tensor)
    boxes: torch.Tensor = field(default_factory=torch.Tensor)
    dets: torch.Tensor = field(default_factory=torch.Tensor)

@dataclass
class Face:
    """Data class for face attributes.

    Attributes:
        indx (int): Index of the face.
        loc (Location): Location of the face in the image.
        dims (Dimensions): Dimensions of the face (height, width).
        tensor (torch.Tensor): Face tensor.
        ratio (float): Ratio of the face area to the image area.
        preds (Dict[str, Prediction]): Predictions of the face given by predictor set.
    """

    indx: int = field(default_factory=int)
    loc: Location = field(default_factory=Location)
    dims: Dimensions = field(default_factory=Dimensions)
    tensor: torch.Tensor = field(default_factory=torch.Tensor)
    ratio: float = field(default_factory=float)
    preds: Dict[str, Prediction] = field(default_factory=dict)

@dataclass
class ImageData:
    """The main data class used for passing data between the different facetorch modules.

    Attributes:
        path_input (str): Path to the input image.
        path_output (str): Path to the output image where the resulting image is saved.
        img (torch.Tensor): Original image tensor used for drawing purposes.
        tensor (torch.Tensor): Processed image tensor.
        dims (Dimensions): Dimensions of the image (height, width).
        det (Detection): Detection data given by the detector.
        faces (List[Face]): List of faces in the image.
        version (str): Version of the facetorch library.

    """

    path_input: str = field(default_factory=str)
    path_output: Optional[str] = field(default_factory=str)
    img: torch.Tensor = field(default_factory=torch.Tensor)
    tensor: torch.Tensor = field(default_factory=torch.Tensor)
    dims: Dimensions = field(default_factory=Dimensions)
    det: Detection = field(default_factory=Detection)
    faces: List[Face] = field(default_factory=list)
    version: str = field(default_factory=str)
    warnings: List[str] = field(default_factory=list)

    def add_preds(
        self,
        preds_list: List[Prediction],
        predictor_name: str,
        face_offset: int = 0,
    ) -> None:
        """Adds a list of predictions to the data object.

        Args:
            preds_list (List[Prediction]): List of predictions.
            predictor_name (str): Name of the predictor.
            face_offset (int): Offset of the face index where the predictions are added.

        Returns:
            None

        """
        j = 0
        for i in range(face_offset, face_offset + len(preds_list)):
            self.faces[i].preds[predictor_name] = preds_list[j]
            j += 1

    def reset_img(self) -> None:
        """Reset the original image tensor to empty state."""
        self.img = torch.tensor([])

    def reset_tensor(self) -> None:
        """Reset the processed image tensor to empty state."""
        self.tensor = torch.tensor([])

    def reset_face_tensors(self) -> None:
        """Reset the face tensors to empty state."""
        for i in range(0, len(self.faces)):
            self.faces[i].tensor = torch.tensor([])

    def reset_face_pred_tensors(self) -> None:
        """Reset prediction tensors while preserving non-tensor metadata."""
        for i in range(0, len(self.faces)):
            for key in self.faces[i].preds:
                prediction = self.faces[i].preds[key]
                prediction.logits = torch.tensor([])
                cleaned_other = _without_tensors(prediction.other)
                prediction.other = (
                    {} if cleaned_other is _REMOVED_TENSOR else cleaned_other
                )

    def reset_det_tensors(self) -> None:
        """Reset the detection object to empty state."""
        self.det = Detection()

    @Timer(
        "ImageData.reset_faces", "{name}: {milliseconds:.2f} ms", logger=logger.debug
    )
    def reset_tensors(self) -> None:
        """Reset the tensors to empty state."""
        self.reset_img()
        self.reset_tensor()
        self.reset_face_tensors()
        self.reset_face_pred_tensors()
        self.reset_det_tensors()

    def set_dims(self) -> None:
        """Set the dimensions attribute from the tensor attribute."""
        self.dims.height = self.tensor.shape[2]
        self.dims.width = self.tensor.shape[3]

    def aggregate_loc_tensor(self) -> torch.Tensor:
        """Aggregates the location tensor from all faces.

        Returns:
            torch.Tensor: Aggregated location tensor for drawing purposes.
        """
        loc_tensor = torch.zeros((len(self.faces), 4), dtype=torch.float32)
        for i in range(0, len(self.faces)):
            loc_tensor[i] = torch.tensor(
                [
                    self.faces[i].loc.x1,
                    self.faces[i].loc.y1,
                    self.faces[i].loc.x2,
                    self.faces[i].loc.y2,
                ]
            )
        return loc_tensor

class InferenceError(FacetorchError, RuntimeError):
    """A configured model failed while executing inference."""

class InputCoercionWarning(UserWarning):
    """A deterministic input conversion was performed in ``coerce`` mode."""

class InputError(FacetorchError, ValueError):
    """The caller supplied an unsupported or ambiguous image input."""

@dataclass(frozen=True)
class InputSpec:
    """Explicitly describes an array or tensor image representation.

    Fields may be omitted in ``coerce`` mode. In ``strict`` mode, callers must
    declare any representation that differs from the source-specific exact
    defaults: uint8 RGB, CHW for Torch, and HWC for NumPy.

    NumPy arrays that look channel-first or are plausible under both conventions
    require an explicit ``CHW``/``BCHW`` or ``HWC``/``BHWC`` layout.
    """

    layout: Optional[InputLayout] = None
    value_range: Optional[InputValueRange] = None
    color_space: Optional[InputColorSpace] = None
    alpha_mode: Optional[AlphaMode] = None

    def __post_init__(self):
        normalized = {}
        for field_name, value, operation in (
            ("layout", self.layout, str.upper),
            ("value_range", self.value_range, str.lower),
            ("color_space", self.color_space, str.upper),
            ("alpha_mode", self.alpha_mode, str.lower),
        ):
            if value is not None and not isinstance(value, str):
                raise InputError(
                    f"InputSpec.{field_name} must be a string or None, "
                    f"got {type(value).__name__}."
                )
            normalized[field_name] = operation(value) if value is not None else None
        for field_name, value in normalized.items():
            object.__setattr__(self, field_name, value)

        valid_values = {
            "layout": {"HW", "CHW", "HWC", "BCHW", "BHWC"},
            "value_range": {"0_1", "0_255"},
            "color_space": {"GRAY", "RGB", "BGR", "RGBA"},
            "alpha_mode": {"drop"},
        }
        for field_name, allowed in valid_values.items():
            value = getattr(self, field_name)
            if value is not None and value not in allowed:
                choices = ", ".join(sorted(allowed))
                raise InputError(
                    f"Invalid InputSpec.{field_name}={value!r}; expected one of {choices}."
                )

class LegacyModelWarning(UserWarning):
    """An explicitly enabled legacy TorchScript artifact was selected."""

@dataclass
class Location:
    """Data class for face location.

    Attributes:
        x1 (int): x1 coordinate
        x2 (int): x2 coordinate
        y1 (int): y1 coordinate
        y2 (int): y2 coordinate
    """

    x1: int = field(default=0)
    x2: int = field(default=0)
    y1: int = field(default=0)
    y2: int = field(default=0)

    def form_square(self) -> None:
        """Form a square from the location.

        Returns:
            None
        """
        height = self.y2 - self.y1
        width = self.x2 - self.x1

        if height > width:
            diff = height - width
            low = diff // 2
            self.x1 -= low
            self.x2 += diff - low
        elif height < width:
            diff = width - height
            low = diff // 2
            self.y1 -= low
            self.y2 += diff - low

    def expand(self, amount: float) -> None:
        """Expand the location while keeping the center.

        Args:
            amount (float): Amount to expand the location by in multiples of the original size.


        Returns:
            None
        """
        if amount < 0:
            raise ValueError("amount must be greater than or equal to 0.")
        if amount != 0.0:
            width = self.x2 - self.x1
            height = self.y2 - self.y1
            expand_x = int(round(width * amount / 2))
            expand_y = int(round(height * amount / 2))
            self.x1 -= expand_x
            self.y1 -= expand_y
            self.x2 += expand_x
            self.y2 += expand_y

    def clamp(self, width: int, height: int) -> None:
        """Clamp coordinates to an image boundary."""
        self.x1 = max(0, min(int(self.x1), int(width)))
        self.x2 = max(self.x1, min(int(self.x2), int(width)))
        self.y1 = max(0, min(int(self.y1), int(height)))
        self.y2 = max(self.y1, min(int(self.y2), int(height)))

    def fit_square(self, width: int, height: int) -> None:
        """Fit the largest possible square around this location inside an image."""
        center_x = (self.x1 + self.x2) / 2.0
        center_y = (self.y1 + self.y2) / 2.0
        side = min(max(self.x2 - self.x1, self.y2 - self.y1), width, height)
        side = max(0, int(round(side)))

        x1 = int(round(center_x - side / 2.0))
        y1 = int(round(center_y - side / 2.0))
        x1 = min(max(0, x1), max(0, width - side))
        y1 = min(max(0, y1), max(0, height - side))
        self.x1, self.y1 = x1, y1
        self.x2, self.y2 = x1 + side, y1 + side

class ModelCompatibilityError(FacetorchError, RuntimeError):
    """No model artifact is compatible with the active runtime."""

class OfflineCacheError(FacetorchError, FileNotFoundError):
    """Offline execution was requested but a required cache entry is absent."""

@dataclass(frozen=True)
class PrefetchItem:
    """One selected artifact and its current verified-cache state."""

    component: str
    artifact_id: str
    path: Path
    format: str
    size_bytes: int
    sha256: str
    cached: bool

@dataclass(frozen=True)
class PrefetchPlan:
    """Download-cost estimate produced before any network request."""

    profile: str
    items: tuple[PrefetchItem, ...]

    @property
    def total_bytes(self) -> int:
        return sum(item.size_bytes for item in self.items)

    @property
    def cached_bytes(self) -> int:
        return sum(item.size_bytes for item in self.items if item.cached)

    @property
    def download_bytes(self) -> int:
        return self.total_bytes - self.cached_bytes

@dataclass(frozen=True)
class PrefetchResult:
    """Completed prefetch result with authenticated local paths."""

    plan: PrefetchPlan
    paths: tuple[Path, ...]

@dataclass
class Prediction:
    """Data class for face prediction results and derivatives.

    Attributes:
        label (str): Label of the face given by predictor.
        logits (torch.Tensor): Output of the predictor model for the face.
        other (Dict): Any other predictions and derivatives for the face.
    """

    label: str = field(default_factory=str)
    logits: torch.Tensor = field(default_factory=torch.Tensor)
    other: Dict = field(default_factory=dict)

@dataclass
class Response:
    """Data class for response data, which is a subset of ImageData.

    Attributes:
        faces (List[Face]): List of faces in the image.
        version (str): Version of the facetorch library.

    """

    faces: List[Face] = field(default_factory=list)
    version: str = field(default_factory=str)