Module facetorch.artifacts

Immutable model-manifest resolution and cache-integrity primitives.

Functions

def detect_model_format(path: str | Path) ‑> str

Expand source code

def detect_model_format(path: str | Path) -> str:
    """Identify modern export versus TorchScript by archive members, without execution."""
    try:
        with ZipFile(path) as archive:
            names = tuple(archive.namelist())
    except (BadZipFile, OSError):
        return "unknown"
    if any(name.endswith("serialized_exported_program.json") for name in names) or (
        any(name.endswith("archive_format") for name in names)
        and any(name.endswith("models/model.json") for name in names)
    ):
        return "pt2"
    if any(name.endswith("data.pkl") for name in names) and any(
        "/code/" in f"/{name}" for name in names
    ):
        return "torchscript"
    return "unknown"

Identify modern export versus TorchScript by archive members, without execution.

def get_model_manifest() ‑> ArtifactManifest

Expand source code

@lru_cache(maxsize=1)
def get_model_manifest() -> ArtifactManifest:
    """Load and validate the immutable manifest packaged with facetorch."""
    model_resources = resources.files("facetorch.models")
    manifest_file = model_resources.joinpath("manifest.json")
    raw = json.loads(manifest_file.read_text(encoding="utf-8"))

    def referenced_json(field: str) -> Mapping[str, Any]:
        filename = str(raw[field])
        if Path(filename).name != filename or not filename.endswith(".json"):
            raise ConfigurationError(
                f"Invalid manifest resource reference: {filename!r}"
            )
        value = json.loads(
            model_resources.joinpath(filename).read_text(encoding="utf-8")
        )
        if not isinstance(value, Mapping):
            raise ConfigurationError(
                f"Manifest resource {filename!r} is not an object."
            )
        return value

    return ArtifactManifest.from_mapping(
        raw,
        compatibility=referenced_json("compatibility_ref"),
        governance=referenced_json("governance_ref"),
    )

Load and validate the immutable manifest packaged with facetorch.

def normalize_device(value: Any) ‑> str

Expand source code

def normalize_device(value: Any) -> str:
    """Normalize torch-style device values to their manifest device family."""
    device = str(value).strip().lower().split(":", 1)[0]
    if device not in {"cpu", "cuda"}:
        raise ConfigurationError(
            f"Model artifacts support only cpu or cuda, got {value!r}."
        )
    return device

Normalize torch-style device values to their manifest device family.

def parse_runtime_version(value: str) ‑> tuple[int, int]

Expand source code

def parse_runtime_version(value: str) -> tuple[int, int]:
    """Parse a runtime version into a comparable major/minor tuple."""
    match = _VERSION_RE.match(str(value).strip())
    if not match:
        raise ConfigurationError(f"Cannot parse runtime version {value!r}.")
    return int(match.group(1)), int(match.group(2))

Parse a runtime version into a comparable major/minor tuple.

def sha256_file(path: str | Path, *, chunk_size: int = 1048576) ‑> str

Expand source code

def sha256_file(path: str | Path, *, chunk_size: int = 1024 * 1024) -> str:
    """Stream a file SHA-256 without loading the artifact into memory."""
    digest = hashlib.sha256()
    with Path(path).open("rb") as artifact_file:
        while chunk := artifact_file.read(chunk_size):
            digest.update(chunk)
    return digest.hexdigest()

Stream a file SHA-256 without loading the artifact into memory.

def verify_artifact(path: str | Path,
descriptor: ArtifactDescriptor) ‑> pathlib.Path

Expand source code

def verify_artifact(path: str | Path, descriptor: ArtifactDescriptor) -> Path:
    """Verify size, digest, and non-executing format before a cache entry is trusted."""
    artifact_path = Path(path)
    try:
        actual_size = artifact_path.stat().st_size
    except OSError as exc:
        raise ArtifactIntegrityError(
            f"Cannot inspect artifact {artifact_path}."
        ) from exc
    if actual_size != descriptor.size_bytes:
        raise ArtifactIntegrityError(
            f"Artifact {descriptor.artifact_id!r} has size {actual_size}; expected "
            f"{descriptor.size_bytes}."
        )
    actual_hash = sha256_file(artifact_path)
    if actual_hash != descriptor.sha256:
        raise ArtifactIntegrityError(
            f"Artifact {descriptor.artifact_id!r} failed SHA-256 verification."
        )
    if descriptor.format in {"pt2", "torchscript"}:
        actual_format = detect_model_format(artifact_path)
        if actual_format != descriptor.format:
            raise ArtifactIntegrityError(
                f"Artifact {descriptor.artifact_id!r} has format {actual_format!r}; "
                f"expected {descriptor.format!r}."
            )
    return artifact_path

Verify size, digest, and non-executing format before a cache entry is trusted.

Classes

class ArtifactDescriptor (artifact_id: str,
model_id: str,
task: str,
source: str,
repo_id: str,
revision: str,
filename: str,
format: str,
artifact_cohort: Optional[str],
sha256: str,
size_bytes: int,
torch_min: Optional[str],
torch_max_exclusive: Optional[str],
devices: tuple[str, ...],
schema_major: Optional[int],
schema_minor: Optional[int],
validation_metadata: Optional[str],
source_weight_sha256: Optional[str],
export_commit: Optional[str],
license_ref: Optional[str],
priority: int = 0)

Expand source code

@dataclass(frozen=True)
class ArtifactDescriptor:
    """One immutable, hash-addressed remote artifact candidate."""

    artifact_id: str
    model_id: str
    task: str
    source: str
    repo_id: str
    revision: str
    filename: str
    format: str
    artifact_cohort: Optional[str]
    sha256: str
    size_bytes: int
    torch_min: Optional[str]
    torch_max_exclusive: Optional[str]
    devices: tuple[str, ...]
    schema_major: Optional[int]
    schema_minor: Optional[int]
    validation_metadata: Optional[str]
    source_weight_sha256: Optional[str]
    export_commit: Optional[str]
    license_ref: Optional[str]
    priority: int = 0

    @classmethod
    def from_mapping(
        cls,
        model_id: str,
        model: Mapping[str, Any],
        raw: Mapping[str, Any],
    ) -> "ArtifactDescriptor":
        try:
            descriptor = cls(
                artifact_id=str(raw["id"]),
                model_id=model_id,
                task=str(model["task"]),
                source=str(model.get("source", "huggingface")),
                repo_id=str(model["repo_id"]),
                revision=str(model["revision"]),
                filename=str(raw["filename"]),
                format=str(raw["format"]),
                artifact_cohort=raw.get("artifact_cohort", raw.get("torch_min")),
                sha256=str(raw["sha256"]).lower(),
                size_bytes=int(raw["size_bytes"]),
                torch_min=raw.get("torch_min"),
                torch_max_exclusive=raw.get("torch_max_exclusive"),
                devices=tuple(str(item).lower() for item in raw["devices"]),
                schema_major=raw.get("schema_major"),
                schema_minor=raw.get("schema_minor"),
                validation_metadata=raw.get("validation_metadata"),
                source_weight_sha256=model.get("source_weight_sha256"),
                export_commit=model.get("export_commit"),
                license_ref=model.get("license_ref"),
                priority=int(raw.get("priority", 0)),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise ConfigurationError(
                f"Invalid artifact descriptor for model {model_id!r}."
            ) from exc
        descriptor.validate()
        return descriptor

    def validate(self) -> None:
        if not self.artifact_id or not self.model_id or not self.task:
            raise ConfigurationError("Artifact identifiers and task cannot be empty.")
        if self.source == "huggingface" and not _REVISION_RE.fullmatch(self.revision):
            raise ConfigurationError(
                f"Artifact {self.artifact_id!r} must pin a 40-character Hub commit."
            )
        if Path(self.filename).name != self.filename:
            raise ConfigurationError(
                f"Artifact filename must be a basename: {self.filename!r}."
            )
        if self.format not in _FORMATS:
            raise ConfigurationError(
                f"Artifact {self.artifact_id!r} has unsupported format {self.format!r}."
            )
        if not _SHA256_RE.fullmatch(self.sha256) or self.size_bytes <= 0:
            raise ConfigurationError(
                f"Artifact {self.artifact_id!r} needs a valid SHA-256 and size."
            )
        if not self.devices or any(
            item not in {"cpu", "cuda"} for item in self.devices
        ):
            raise ConfigurationError(
                f"Artifact {self.artifact_id!r} has invalid device eligibility."
            )
        lower = _version_bound(self.torch_min)
        upper = _version_bound(self.torch_max_exclusive)
        if lower is not None and upper is not None and lower >= upper:
            raise ConfigurationError(
                f"Artifact {self.artifact_id!r} has an empty runtime range."
            )
        if self.format == "pt2" and not self.filename.endswith(".pt2"):
            raise ConfigurationError(
                f"Export artifact {self.artifact_id!r} must preserve .pt2."
            )
        if self.format == "pt2" and (
            (self.torch_min is not None or self.torch_max_exclusive is not None)
            and (
                self.artifact_cohort is None
                or _VERSION_RE.fullmatch(str(self.artifact_cohort)) is None
            )
        ):
            raise ConfigurationError(
                f"Export artifact {self.artifact_id!r} needs a canonical cohort."
            )
        if self.format == "torchscript" and not self.filename.endswith(".pt"):
            raise ConfigurationError(
                f"TorchScript artifact {self.artifact_id!r} must preserve .pt."
            )

    def supports(self, runtime: tuple[int, int], device: str) -> bool:
        lower = _version_bound(self.torch_min)
        upper = _version_bound(self.torch_max_exclusive)
        return (
            device in self.devices
            and (lower is None or runtime >= lower)
            and (upper is None or runtime < upper)
        )

    def cache_path(self, configured_path: str | Path) -> Path:
        """Place each candidate under its authenticated real filename."""
        return Path(configured_path).expanduser().parent / self.filename

One immutable, hash-addressed remote artifact candidate.

Static methods

def from_mapping(model_id: str, model: Mapping[str, Any], raw: Mapping[str, Any]) ‑> ArtifactDescriptor

Instance variables

var artifact_id : str

var model_id : str

var task : str

var source : str

var repo_id : str

var revision : str

var filename : str

var format : str

var artifact_cohort : str | None

var sha256 : str

var size_bytes : int

var torch_min : str | None

var torch_max_exclusive : str | None

var devices : tuple[str, ...]

var schema_major : int | None

var schema_minor : int | None

var validation_metadata : str | None

var source_weight_sha256 : str | None

var export_commit : str | None

var license_ref : str | None

var priority : int

Methods

def validate(self) ‑> None

Expand source code

def validate(self) -> None:
    if not self.artifact_id or not self.model_id or not self.task:
        raise ConfigurationError("Artifact identifiers and task cannot be empty.")
    if self.source == "huggingface" and not _REVISION_RE.fullmatch(self.revision):
        raise ConfigurationError(
            f"Artifact {self.artifact_id!r} must pin a 40-character Hub commit."
        )
    if Path(self.filename).name != self.filename:
        raise ConfigurationError(
            f"Artifact filename must be a basename: {self.filename!r}."
        )
    if self.format not in _FORMATS:
        raise ConfigurationError(
            f"Artifact {self.artifact_id!r} has unsupported format {self.format!r}."
        )
    if not _SHA256_RE.fullmatch(self.sha256) or self.size_bytes <= 0:
        raise ConfigurationError(
            f"Artifact {self.artifact_id!r} needs a valid SHA-256 and size."
        )
    if not self.devices or any(
        item not in {"cpu", "cuda"} for item in self.devices
    ):
        raise ConfigurationError(
            f"Artifact {self.artifact_id!r} has invalid device eligibility."
        )
    lower = _version_bound(self.torch_min)
    upper = _version_bound(self.torch_max_exclusive)
    if lower is not None and upper is not None and lower >= upper:
        raise ConfigurationError(
            f"Artifact {self.artifact_id!r} has an empty runtime range."
        )
    if self.format == "pt2" and not self.filename.endswith(".pt2"):
        raise ConfigurationError(
            f"Export artifact {self.artifact_id!r} must preserve .pt2."
        )
    if self.format == "pt2" and (
        (self.torch_min is not None or self.torch_max_exclusive is not None)
        and (
            self.artifact_cohort is None
            or _VERSION_RE.fullmatch(str(self.artifact_cohort)) is None
        )
    ):
        raise ConfigurationError(
            f"Export artifact {self.artifact_id!r} needs a canonical cohort."
        )
    if self.format == "torchscript" and not self.filename.endswith(".pt"):
        raise ConfigurationError(
            f"TorchScript artifact {self.artifact_id!r} must preserve .pt."
        )

def supports(self, runtime: tuple[int, int], device: str) ‑> bool

Expand source code

def supports(self, runtime: tuple[int, int], device: str) -> bool:
    lower = _version_bound(self.torch_min)
    upper = _version_bound(self.torch_max_exclusive)
    return (
        device in self.devices
        and (lower is None or runtime >= lower)
        and (upper is None or runtime < upper)
    )

def cache_path(self, configured_path: str | Path) ‑> pathlib.Path

Expand source code

def cache_path(self, configured_path: str | Path) -> Path:
    """Place each candidate under its authenticated real filename."""
    return Path(configured_path).expanduser().parent / self.filename

Place each candidate under its authenticated real filename.

class ArtifactManifest (*,
manifest_version: int,
manifest_revision: str,
status: str,
models: Mapping[str, tuple[ArtifactDescriptor, ...]],
compatibility_status: str = 'unspecified',
torch_specifier: str = '',
python_specifier: str = '',
supported_torch_minors: tuple[str, ...] = (),
required_devices: tuple[str, ...] = (),
governance_status: str = 'unspecified',
governance: Mapping[str, ModelGovernance] | None = None)

Expand source code

class ArtifactManifest:
    """Validated in-memory view of the packaged model manifest."""

    def __init__(
        self,
        *,
        manifest_version: int,
        manifest_revision: str,
        status: str,
        models: Mapping[str, tuple[ArtifactDescriptor, ...]],
        compatibility_status: str = "unspecified",
        torch_specifier: str = "",
        python_specifier: str = "",
        supported_torch_minors: tuple[str, ...] = (),
        required_devices: tuple[str, ...] = (),
        governance_status: str = "unspecified",
        governance: Mapping[str, ModelGovernance] | None = None,
    ) -> None:
        if manifest_version != 1:
            raise ConfigurationError(
                f"Unsupported model manifest version {manifest_version!r}."
            )
        if not manifest_revision or status not in _MANIFEST_STATUSES:
            raise ConfigurationError("Model manifest metadata is incomplete.")
        self.manifest_version = manifest_version
        self.manifest_revision = manifest_revision
        self.status = status
        self.models = dict(models)
        self.compatibility_status = compatibility_status
        self.torch_specifier = torch_specifier
        self.python_specifier = python_specifier
        self.supported_torch_minors = supported_torch_minors
        self.required_devices = required_devices
        self.governance_status = governance_status
        self.governance = dict(governance or {})
        self._by_artifact_id = {
            artifact.artifact_id: artifact
            for artifacts in self.models.values()
            for artifact in artifacts
        }
        expected = sum(len(items) for items in self.models.values())
        if len(self._by_artifact_id) != expected:
            raise ConfigurationError("Artifact IDs must be globally unique.")
        if self.supported_torch_minors:
            for model_id, artifacts in self.models.items():
                for minor in self.supported_torch_minors:
                    runtime = parse_runtime_version(minor)
                    matching = [
                        artifact
                        for artifact in artifacts
                        if artifact.format == "pt2"
                        and all(
                            artifact.supports(runtime, device)
                            for device in self.required_devices
                        )
                    ]
                    if len(matching) != 1:
                        raise ConfigurationError(
                            f"Model {model_id!r} needs exactly one artifact for "
                            f"supported torch {minor}; found {len(matching)}."
                        )
        if status == "approved":
            incomplete = [
                artifact.artifact_id
                for artifact in self._by_artifact_id.values()
                if not artifact.source_weight_sha256
                or not _SHA256_RE.fullmatch(artifact.source_weight_sha256)
                or not artifact.export_commit
                or not _REVISION_RE.fullmatch(artifact.export_commit)
                or (artifact.format == "pt2" and not artifact.validation_metadata)
                or not artifact.license_ref
            ]
            governance_incomplete = sorted(
                model_id
                for model_id in self.models
                if model_id not in self.governance
                or not self.governance[model_id].approved
            )
            if (
                incomplete
                or governance_incomplete
                or self.compatibility_status != "approved"
                or self.governance_status != "approved"
            ):
                raise ConfigurationError(
                    "An approved manifest requires complete provenance, validation, "
                    "rights, and compatibility metadata; incomplete artifacts: "
                    f"{', '.join(incomplete) or 'none'}; incomplete models: "
                    f"{', '.join(governance_incomplete) or 'none'}."
                )

    @classmethod
    def from_mapping(
        cls,
        raw: Mapping[str, Any],
        *,
        compatibility: Mapping[str, Any] | None = None,
        governance: Mapping[str, Any] | None = None,
    ) -> "ArtifactManifest":
        try:
            raw_models = raw["models"]
            models = {
                model_id: tuple(
                    ArtifactDescriptor.from_mapping(model_id, model, artifact)
                    for artifact in model["artifacts"]
                )
                for model_id, model in raw_models.items()
            }
            if not models or any(not artifacts for artifacts in models.values()):
                raise ConfigurationError("Every manifest model needs an artifact.")

            compatibility_status = "unspecified"
            torch_specifier = ""
            python_specifier = ""
            supported_torch_minors: tuple[str, ...] = ()
            required_devices: tuple[str, ...] = ()
            if compatibility is not None:
                compatibility_schema = int(compatibility["schema_version"])
                if compatibility_schema not in {1, 2}:
                    raise ConfigurationError("Unsupported compatibility schema.")
                compatibility_status = str(compatibility["status"])
                if compatibility_status not in _COMPATIBILITY_STATUSES:
                    raise ConfigurationError("Invalid compatibility status.")
                torch_record = compatibility["torch"]
                python_record = compatibility["python"]
                platform_policy = compatibility["platform_policy"]
                torch_specifier = str(torch_record["specifier"])
                python_specifier = str(python_record["specifier"])
                supported_torch_minors = tuple(
                    str(item) for item in torch_record["supported_minor_lines"]
                )
                required_devices = tuple(
                    str(item) for item in platform_policy["required_devices"]
                )
                if (
                    not supported_torch_minors
                    or len(set(supported_torch_minors)) != len(supported_torch_minors)
                    or not required_devices
                    or any(item not in {"cpu", "cuda"} for item in required_devices)
                ):
                    raise ConfigurationError("Compatibility matrix is incomplete.")
                if compatibility_schema == 2:
                    cohort_records = compatibility.get("cohorts")
                    runtime_lanes = compatibility.get("runtime_lanes")
                    if not isinstance(cohort_records, list) or not isinstance(
                        runtime_lanes, list
                    ):
                        raise ConfigurationError(
                            "Compatibility matrix needs artifact cohorts and runtime lanes."
                        )
                    artifact_cohorts = {
                        str(record.get("artifact_cohort"))
                        for record in cohort_records
                        if isinstance(record, Mapping)
                    }
                    lane_minors = [
                        str(record.get("torch_minor"))
                        for record in runtime_lanes
                        if isinstance(record, Mapping)
                    ]
                    if (
                        len(artifact_cohorts) != len(cohort_records)
                        or len(lane_minors) != len(runtime_lanes)
                        or len(set(lane_minors)) != len(lane_minors)
                        or set(lane_minors) != set(supported_torch_minors)
                    ):
                        raise ConfigurationError(
                            "Compatibility artifact cohorts or runtime lanes are inconsistent."
                        )
                    cohort_ranges = {}
                    for record in cohort_records:
                        cohort = str(record.get("artifact_cohort"))
                        lower = parse_runtime_version(str(record.get("torch_min")))
                        upper = parse_runtime_version(
                            str(record.get("torch_max_exclusive"))
                        )
                        if lower >= upper:
                            raise ConfigurationError(
                                f"Compatibility artifact cohort {cohort!r} has an empty range."
                            )
                        cohort_ranges[cohort] = (lower, upper)
                    for record in runtime_lanes:
                        lane = str(record.get("torch_minor"))
                        cohort = str(record.get("artifact_cohort"))
                        runtime = parse_runtime_version(lane)
                        bounds = cohort_ranges.get(cohort)
                        if bounds is None or not bounds[0] <= runtime < bounds[1]:
                            raise ConfigurationError(
                                f"Compatibility runtime lane {lane!r} has no valid artifact route."
                            )

            governance_status = "unspecified"
            governance_records: dict[str, ModelGovernance] = {}
            if governance is not None:
                if int(governance["schema_version"]) != 1:
                    raise ConfigurationError("Unsupported governance schema.")
                governance_status = str(governance["status"])
                if governance_status not in _GOVERNANCE_STATUSES:
                    raise ConfigurationError("Invalid governance status.")
                governance_records = {
                    model_id: ModelGovernance.from_mapping(model_id, record)
                    for model_id, record in governance["models"].items()
                }
                if set(governance_records) != set(models):
                    raise ConfigurationError(
                        "Governance records must exactly cover manifest models."
                    )
            return cls(
                manifest_version=int(raw["manifest_version"]),
                manifest_revision=str(raw["manifest_revision"]),
                status=str(raw["status"]),
                models=models,
                compatibility_status=compatibility_status,
                torch_specifier=torch_specifier,
                python_specifier=python_specifier,
                supported_torch_minors=supported_torch_minors,
                required_devices=required_devices,
                governance_status=governance_status,
                governance=governance_records,
            )
        except ConfigurationError:
            raise
        except (AttributeError, KeyError, TypeError, ValueError) as exc:
            raise ConfigurationError("Invalid model manifest structure.") from exc

    @classmethod
    def from_json(cls, value: str) -> "ArtifactManifest":
        try:
            raw = json.loads(value)
        except json.JSONDecodeError as exc:
            raise ConfigurationError("Model manifest is not valid JSON.") from exc
        if not isinstance(raw, Mapping):
            raise ConfigurationError("Model manifest root must be an object.")
        return cls.from_mapping(raw)

    def descriptor(self, artifact_id: str) -> ArtifactDescriptor:
        try:
            return self._by_artifact_id[artifact_id]
        except KeyError as exc:
            raise ConfigurationError(f"Unknown artifact ID {artifact_id!r}.") from exc

    def candidates(
        self,
        model_id: str,
        *,
        torch_version: str,
        device: Any,
        allow_legacy_models: bool = False,
        incompatible_artifact_ids: Iterable[str] = (),
    ) -> tuple[ArtifactDescriptor, ...]:
        try:
            artifacts = self.models[model_id]
        except KeyError as exc:
            raise ConfigurationError(f"Unknown manifest model {model_id!r}.") from exc
        runtime = parse_runtime_version(torch_version)
        device_family = normalize_device(device)
        runtime_label = f"{runtime[0]}.{runtime[1]}"
        if (
            self.supported_torch_minors
            and runtime_label not in self.supported_torch_minors
        ):
            supported = ", ".join(self.supported_torch_minors)
            raise ModelCompatibilityError(
                f"Torch {runtime_label} is outside facetorch's supported minor "
                f"lines ({supported}); no model download was attempted."
            )
        exports = [
            item
            for item in artifacts
            if item.format == "pt2" and item.supports(runtime, device_family)
        ]
        legacy = [
            item
            for item in artifacts
            if item.format == "torchscript" and item.supports(runtime, device_family)
        ]
        exports.sort(key=lambda item: item.priority)
        legacy.sort(key=lambda item: item.priority)
        selected = exports + (legacy if allow_legacy_models else [])
        if selected:
            if isinstance(incompatible_artifact_ids, (str, bytes)):
                raise ConfigurationError(
                    "incompatible_artifact_ids must be a collection of artifact IDs."
                )
            rejected = set(incompatible_artifact_ids)
            if any(not isinstance(item, str) or not item for item in rejected):
                raise ConfigurationError(
                    "incompatible_artifact_ids must contain non-empty strings."
                )
            eligible = tuple(
                item for item in selected if item.artifact_id not in rejected
            )
            if eligible:
                return eligible
            raise ModelCompatibilityError(
                f"All eligible artifacts for {model_id!r} were already rejected by "
                f"torch {runtime_label} on {device_family}."
            )

        if legacy and not allow_legacy_models:
            remedy = " Set allow_legacy_models=True to use an eligible verified legacy artifact."
        else:
            remedy = (
                " Install a documented supported runtime or publish a validated cohort."
            )
        raise ModelCompatibilityError(
            f"No compatible artifact for {model_id!r} with torch {runtime_label} "
            f"on {device_family}.{remedy}"
        )

    def iter_descriptors(self) -> Iterable[ArtifactDescriptor]:
        for model_id in sorted(self.models):
            yield from self.models[model_id]

Validated in-memory view of the packaged model manifest.

Static methods

def from_mapping(raw: Mapping[str, Any],
*,
compatibility: Mapping[str, Any] | None = None,
governance: Mapping[str, Any] | None = None) ‑> ArtifactManifest

def from_json(value: str) ‑> ArtifactManifest

Methods

def descriptor(self, artifact_id: str) ‑> ArtifactDescriptor

Expand source code

def descriptor(self, artifact_id: str) -> ArtifactDescriptor:
    try:
        return self._by_artifact_id[artifact_id]
    except KeyError as exc:
        raise ConfigurationError(f"Unknown artifact ID {artifact_id!r}.") from exc

def candidates(self,
model_id: str,
*,
torch_version: str,
device: Any,
allow_legacy_models: bool = False,
incompatible_artifact_ids: Iterable[str] = ()) ‑> tuple[ArtifactDescriptor, ...]

Expand source code

def candidates(
    self,
    model_id: str,
    *,
    torch_version: str,
    device: Any,
    allow_legacy_models: bool = False,
    incompatible_artifact_ids: Iterable[str] = (),
) -> tuple[ArtifactDescriptor, ...]:
    try:
        artifacts = self.models[model_id]
    except KeyError as exc:
        raise ConfigurationError(f"Unknown manifest model {model_id!r}.") from exc
    runtime = parse_runtime_version(torch_version)
    device_family = normalize_device(device)
    runtime_label = f"{runtime[0]}.{runtime[1]}"
    if (
        self.supported_torch_minors
        and runtime_label not in self.supported_torch_minors
    ):
        supported = ", ".join(self.supported_torch_minors)
        raise ModelCompatibilityError(
            f"Torch {runtime_label} is outside facetorch's supported minor "
            f"lines ({supported}); no model download was attempted."
        )
    exports = [
        item
        for item in artifacts
        if item.format == "pt2" and item.supports(runtime, device_family)
    ]
    legacy = [
        item
        for item in artifacts
        if item.format == "torchscript" and item.supports(runtime, device_family)
    ]
    exports.sort(key=lambda item: item.priority)
    legacy.sort(key=lambda item: item.priority)
    selected = exports + (legacy if allow_legacy_models else [])
    if selected:
        if isinstance(incompatible_artifact_ids, (str, bytes)):
            raise ConfigurationError(
                "incompatible_artifact_ids must be a collection of artifact IDs."
            )
        rejected = set(incompatible_artifact_ids)
        if any(not isinstance(item, str) or not item for item in rejected):
            raise ConfigurationError(
                "incompatible_artifact_ids must contain non-empty strings."
            )
        eligible = tuple(
            item for item in selected if item.artifact_id not in rejected
        )
        if eligible:
            return eligible
        raise ModelCompatibilityError(
            f"All eligible artifacts for {model_id!r} were already rejected by "
            f"torch {runtime_label} on {device_family}."
        )

    if legacy and not allow_legacy_models:
        remedy = " Set allow_legacy_models=True to use an eligible verified legacy artifact."
    else:
        remedy = (
            " Install a documented supported runtime or publish a validated cohort."
        )
    raise ModelCompatibilityError(
        f"No compatible artifact for {model_id!r} with torch {runtime_label} "
        f"on {device_family}.{remedy}"
    )

def iter_descriptors(self) ‑> Iterable[ArtifactDescriptor]

Expand source code

def iter_descriptors(self) -> Iterable[ArtifactDescriptor]:
    for model_id in sorted(self.models):
        yield from self.models[model_id]