Module facetorch.model_cache

Public planning, prefetch, inspection, and cache-recovery APIs.

Functions

def cleanup_quarantined_cache(root: Optional[str | os.PathLike] = None, *, confirm: bool = False) ‑> CacheCleanupReport

Expand source code

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,
    )

Delete only reported quarantine files and only after explicit confirmation.

def inspect_incompatible_cache(root: Optional[str | os.PathLike] = None) ‑> CacheCleanupReport

Expand source code

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,
    )

Report persisted runtime/schema rejections without changing the cache.

def inspect_legacy_cache(path: str | os.PathLike) ‑> tuple[CacheEntryInspection, ...]

Expand source code

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)

Hash and classify old model files without deserializing or executing them.

def inspect_quarantined_cache(root: Optional[str | os.PathLike] = None) ‑> CacheCleanupReport

Expand source code

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,
    )

Report quarantined entries and reclaimable bytes without deleting anything.

def migrate_legacy_artifact(source: str | os.PathLike, artifact_id: str, destination: str | os.PathLike) ‑> pathlib.Path

Expand source code

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)

Copy one exact manifest match into v1 layout without changing the source.

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

Expand source code

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

Resolve exact artifacts and costs without creating files or using the network.

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

Expand source code

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

Download exactly a planned selection after explicit bulk-cost confirmation.

def reset_incompatible_cache(root: Optional[str | os.PathLike] = None, *, confirm: bool = False) ‑> CacheCleanupReport

Expand source code

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,
    )

Explicitly clear persisted runtime/schema rejections after remediation.

Classes

class CacheCleanupReport (paths: tuple[Path, ...], total_bytes: int, deleted: bool)

Expand source code

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

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

Quarantine inventory and optional explicit cleanup result.

Instance variables

var paths : tuple[pathlib.Path, ...]

var total_bytes : int

var deleted : bool

class CacheEntryInspection (path: Path, size_bytes: int, sha256: str, detected_format: str, mislabeled: bool)

Expand source code

@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

Non-executing inspection result for one possible legacy artifact.

Instance variables

var path : pathlib.Path

var size_bytes : int

var sha256 : str

var detected_format : str

var mislabeled : bool

class PrefetchItem (component: str,
artifact_id: str,
path: Path,
format: str,
size_bytes: int,
sha256: str,
cached: bool)

Expand source code

@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

One selected artifact and its current verified-cache state.

Instance variables

var component : str

var artifact_id : str

var path : pathlib.Path

var format : str

var size_bytes : int

var sha256 : str

var cached : bool

class PrefetchPlan (profile: str,
items: tuple[PrefetchItem, ...])

Expand source code

@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

Download-cost estimate produced before any network request.

Instance variables

var profile : str

var items : tuple[PrefetchItem, ...]

prop total_bytes : int

Expand source code

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

prop cached_bytes : int

Expand source code

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

prop download_bytes : int

Expand source code

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

class PrefetchResult (plan: PrefetchPlan,
paths: tuple[Path, ...])

Expand source code

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

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

Completed prefetch result with authenticated local paths.

Instance variables

var plan : PrefetchPlan

var paths : tuple[pathlib.Path, ...]