Spaces:
Paused
Paused
File size: 2,444 Bytes
1d0f191 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """Fetch the pinned model weights and verify every hash.
Refuses to continue if a downloaded file does not match the lockfile. A model
that quietly differs from the recorded sha would invalidate every number in the
report while still running, which is the worst possible failure mode.
"""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
os.environ.setdefault("HF_HOME", str(Path(__file__).resolve().parents[1] / ".hf_home"))
from forgesight.settings import get_settings
from forgesight.vision.export_onnx import file_sha256
def fetch(force: bool = False) -> int:
from huggingface_hub import hf_hub_download
s = get_settings()
lock_path = Path(__file__).resolve().parents[1] / "models.lock.json"
lock = json.loads(lock_path.read_text())
failures = 0
for name, entry in lock["models"].items():
target = s.models_dir / name
target.mkdir(parents=True, exist_ok=True)
print(f"== {name} @ {entry['revision'][:12]} ({entry['repo_id']})")
for fname, meta in entry["files"].items():
dest = target / fname
if not force and dest.exists() and file_sha256(dest) == meta["sha256"]:
print(f" {fname:28s} ok (already present)")
continue
# revision= is a commit sha, so a moving branch cannot change what we get.
got = hf_hub_download(
repo_id=entry["repo_id"],
filename=fname,
revision=entry["revision"],
cache_dir=s.hf_home / "hub",
)
src = Path(got)
actual = file_sha256(src)
if actual != meta["sha256"]:
print(
f" {fname:28s} SHA MISMATCH\n"
f" expected {meta['sha256']}\n"
f" actual {actual}",
file=sys.stderr,
)
failures += 1
continue
dest.write_bytes(src.read_bytes())
print(f" {fname:28s} verified {actual[:12]}")
return failures
def main() -> int:
force = "--force" in sys.argv
failures = fetch(force=force)
if failures:
print(f"\n{failures} artifact(s) failed verification", file=sys.stderr)
return 1
print("\nall pinned artifacts present and verified")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|