"""Public cache identity and settings-addressed naming helpers."""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from ufp.core._disk_cache import (
cache_settings_digest as _cache_settings_digest,
)
from ufp.core._disk_cache import (
cache_settings_name as _cache_settings_name,
)
[docs]
def normalize_cache_settings(value: object) -> object:
"""Return a JSON-friendly representation for cache identity settings."""
if value is None or isinstance(value, (bool, int, float, str)):
return value
if isinstance(value, Path):
return str(value)
if isinstance(value, Mapping):
return {str(key): normalize_cache_settings(item) for key, item in value.items()}
if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
return [normalize_cache_settings(item) for item in value]
text = str(value)
return text.removeprefix("torch.")
[docs]
@dataclass(frozen=True)
class CacheIdentity:
"""Stable identity for a settings-addressed cache family."""
cache_kind: str
prefix: str
settings: Mapping[str, object]
digest_length: int = 16
@property
def normalized_settings(self) -> dict[str, object]:
"""Return JSON-friendly settings used to compute this cache identity."""
normalized = normalize_cache_settings(self.settings)
if not isinstance(normalized, dict):
raise TypeError("`CacheIdentity.settings` must normalize to a mapping")
return normalized
@property
def digest(self) -> str:
"""Return the stable digest for this cache identity."""
return _cache_settings_digest(
self.normalized_settings,
length=int(self.digest_length),
)
@property
def name(self) -> str:
"""Return the deterministic directory name for this cache identity."""
return _cache_settings_name(
self.prefix,
self.normalized_settings,
length=int(self.digest_length),
)
[docs]
def directory(self, parent: Path | str) -> Path:
"""Return this identity's cache directory under ``parent``."""
return Path(parent) / self.name
[docs]
def summary(
self, *, extra: Mapping[str, object] | None = None
) -> dict[str, object]:
"""Return a JSON-friendly summary suitable for ``cache_settings.json``."""
payload: dict[str, object] = {
"schema_version": 1,
"cache_kind": str(self.cache_kind),
"cache_name": self.name,
"cache_prefix": self.name.removesuffix(f"_{self.digest}"),
"settings_digest": self.digest,
"digest_algorithm": "sha256-json-v1",
"digest_length": int(self.digest_length),
"invalidating_settings": self.normalized_settings,
}
if extra:
payload.update(
{
str(key): normalize_cache_settings(value)
for key, value in extra.items()
}
)
return payload
[docs]
def cache_settings_digest(settings: object, *, length: int = 16) -> str:
"""Return a stable digest for JSON-friendly cache identity settings."""
return _cache_settings_digest(normalize_cache_settings(settings), length=length)
[docs]
def suggest_cache_digest(settings: object, *, length: int = 16) -> str:
"""Return the settings digest used by suggested cache names."""
return cache_settings_digest(settings, length=length)
[docs]
def suggest_cache_name(
prefix: str,
settings: object,
*,
length: int = 16,
) -> str:
"""Return a deterministic cache directory name from settings."""
return _cache_settings_name(
prefix, normalize_cache_settings(settings), length=length
)
[docs]
def suggest_cache_directory(
parent: Path | str,
prefix: str,
settings: object,
*,
length: int = 16,
) -> Path:
"""Return a deterministic cache directory path from settings."""
return Path(parent) / suggest_cache_name(prefix, settings, length=length)
[docs]
def cache_settings_from_names(
namespace: Mapping[str, object],
names: Sequence[str],
*,
example: str,
cache_kind: str,
extra: Mapping[str, object] | None = None,
) -> dict[str, object]:
"""Build a JSON-friendly cache identity payload from named parameters."""
settings: dict[str, object] = {
"example": str(example),
"cache_kind": str(cache_kind),
}
for name in names:
settings[name.lower()] = normalize_cache_settings(namespace[name])
if extra:
settings.update(
{str(key): normalize_cache_settings(value) for key, value in extra.items()}
)
return settings
__all__ = [
"CacheIdentity",
"cache_settings_digest",
"cache_settings_from_names",
"normalize_cache_settings",
"suggest_cache_digest",
"suggest_cache_directory",
"suggest_cache_name",
]