Buckets:
| """cwm3/baselines_dms/registry.py — JSON registry of every third-party weight / package used by the DMS baselines. | |
| Registry file: baselines_dms/weights/registry.json (schema 1) | |
| {"schema": 1, "generated_utc": ..., "entries": [ENTRY, ...]} | |
| ENTRY = {id, kind: pip|url|hf|gdown|git, name, version, url, local_path, sha256, size_bytes, license, license_url, | |
| installed_utc, notes, | |
| files: [{name, url, sha256, size_bytes}] # optional; multi-file entries (xml+bin, OpenFace folder). For a | |
| # multi-file entry `sha256` is the MANIFEST digest (see manifest_sha256) | |
| source: {...}} # kind-specific install hints (hf repo_id/filename/revision, | |
| # gdown folder id, pip_args such as ["--no-deps"], git commit) | |
| Weights live under baselines_dms/weights/<id>/ (local_path is relative to the cwm3 root). | |
| CLI | |
| python -m baselines_dms.registry --list | |
| python -m baselines_dms.registry --verify # sha256 / size / installed-version check of every entry | |
| python -m baselines_dms.registry --install <id>|all # (re)download / pip-install what is missing, then verify | |
| Library | |
| from baselines_dms.registry import resolve; p = resolve("omz_face_detection_adas_0001") # asserts sha256 first | |
| Rules honoured here: never prints secrets (no API keys are used anywhere: every source is public); never writes outside | |
| baselines_dms/; never touches results/. Entries are created by registry_seed.py (the canonical provenance record). | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import importlib.metadata as _md | |
| import json | |
| import shutil | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import time | |
| import urllib.request | |
| from pathlib import Path | |
| HERE = Path(__file__).resolve().parent | |
| CWM3 = HERE.parent | |
| WEIGHTS = HERE / "weights" | |
| REGISTRY_JSON = WEIGHTS / "registry.json" | |
| PYTHON = sys.executable | |
| KINDS = ("pip", "url", "hf", "gdown", "git") | |
| FIELDS = ("id", "kind", "name", "version", "url", "local_path", "sha256", "size_bytes", "license", "license_url", | |
| "installed_utc", "notes") | |
| # ----------------------------------------------------------------------------------------------------------- helpers | |
| def utc_now() -> str: | |
| return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| def sha256_file(path: Path, chunk: int = 1 << 22) -> str: | |
| h = hashlib.sha256() | |
| with open(path, "rb") as f: | |
| for b in iter(lambda: f.read(chunk), b""): | |
| h.update(b) | |
| return h.hexdigest() | |
| def manifest_sha256(files: list[dict]) -> str: | |
| """Digest of the sorted `sha256 name` lines — one hash that pins a multi-file entry.""" | |
| lines = sorted(f"{f['sha256']} {f['name']}" for f in files) | |
| return hashlib.sha256(("\n".join(lines) + "\n").encode()).hexdigest() | |
| def load() -> dict: | |
| if not REGISTRY_JSON.exists(): | |
| return {"schema": 1, "generated_utc": None, "entries": []} | |
| return json.loads(REGISTRY_JSON.read_text()) | |
| def save(reg: dict) -> None: | |
| reg["generated_utc"] = utc_now() | |
| WEIGHTS.mkdir(parents=True, exist_ok=True) | |
| tmp = REGISTRY_JSON.with_suffix(".json.tmp") | |
| tmp.write_text(json.dumps(reg, indent=1, ensure_ascii=False) + "\n") | |
| tmp.replace(REGISTRY_JSON) | |
| def entries() -> dict[str, dict]: | |
| return {e["id"]: e for e in load()["entries"]} | |
| def get(entry_id: str) -> dict: | |
| es = entries() | |
| if entry_id not in es: | |
| raise KeyError(f"unknown registry id {entry_id!r}; known: {sorted(es)}") | |
| return es[entry_id] | |
| def local_path(entry: dict) -> Path: | |
| p = Path(entry["local_path"]) | |
| return p if p.is_absolute() else CWM3 / p | |
| def file_records(entry: dict) -> list[dict]: | |
| """Per-file expectations; a single-file entry is described by local_path + sha256 + size_bytes.""" | |
| if entry.get("files"): | |
| return entry["files"] | |
| if entry["kind"] in ("pip", "git"): | |
| return [] | |
| return [{"name": local_path(entry).name, "url": entry.get("url"), "sha256": entry["sha256"], "size_bytes": entry.get("size_bytes")}] | |
| def file_dir(entry: dict) -> Path: | |
| lp = local_path(entry) | |
| return lp if entry.get("files") else lp.parent | |
| # ------------------------------------------------------------------------------------------------------------ verify | |
| def _installed_version(dist: str) -> str | None: | |
| try: | |
| return _md.version(dist) | |
| except _md.PackageNotFoundError: | |
| return None | |
| def _installed_git_commit(dist: str) -> str | None: | |
| try: | |
| d = _md.distribution(dist) | |
| raw = d.read_text("direct_url.json") | |
| return json.loads(raw)["vcs_info"]["commit_id"] if raw else None | |
| except Exception: | |
| return None | |
| def verify(entry: dict) -> list[str]: | |
| """Return a list of problems (empty == OK). Files: existence + size + sha256. pip: version. git: commit.""" | |
| problems: list[str] = [] | |
| if entry["kind"] == "pip": | |
| v = _installed_version(entry["name"]) | |
| if v is None: | |
| problems.append(f"{entry['name']} not installed") | |
| elif v != entry["version"]: | |
| problems.append(f"{entry['name']} installed {v} != registry {entry['version']}") | |
| return problems | |
| if entry["kind"] == "git": | |
| c = _installed_git_commit(entry["name"]) | |
| if c is None: | |
| problems.append(f"{entry['name']} not installed from git") | |
| elif not c.startswith(entry["version"]): | |
| problems.append(f"{entry['name']} commit {c[:12]} != registry {entry['version'][:12]}") | |
| return problems | |
| d = file_dir(entry) | |
| for f in file_records(entry): | |
| p = d / f["name"] | |
| if not p.exists(): | |
| problems.append(f"missing {p}") | |
| continue | |
| n = p.stat().st_size | |
| if f.get("size_bytes") is not None and n != f["size_bytes"]: | |
| problems.append(f"size {n} != {f['size_bytes']} for {p.name}") | |
| h = sha256_file(p) | |
| if h != f["sha256"]: | |
| problems.append(f"sha256 mismatch for {p.name}: {h[:16]}… != {f['sha256'][:16]}…") | |
| if entry.get("files") and entry.get("sha256") and manifest_sha256(entry["files"]) != entry["sha256"]: | |
| problems.append("manifest sha256 does not match the files list") | |
| return problems | |
| def resolve(entry_id: str) -> Path: | |
| """Path of the entry's file (single-file) or directory (multi-file) — sha256 of EVERY file asserted first.""" | |
| entry = get(entry_id) | |
| problems = verify(entry) | |
| assert not problems, f"registry entry {entry_id!r} failed verification: " + "; ".join(problems) | |
| if entry["kind"] in ("pip", "git"): | |
| raise ValueError(f"{entry_id} is a {entry['kind']} package, not a file; import {entry['name']!r} instead") | |
| return local_path(entry) | |
| # ----------------------------------------------------------------------------------------------------------- install | |
| def _download(url: str, dst: Path) -> None: | |
| dst.parent.mkdir(parents=True, exist_ok=True) | |
| tmp = dst.with_suffix(dst.suffix + ".part") | |
| req = urllib.request.Request(url, headers={"User-Agent": "cwm3-baselines_dms/1.0"}) | |
| with urllib.request.urlopen(req, timeout=120) as r, open(tmp, "wb") as f: | |
| shutil.copyfileobj(r, f, 1 << 20) | |
| tmp.replace(dst) | |
| def _pip(args: list[str]) -> None: | |
| cmd = [PYTHON, "-m", "pip", "install", *args] | |
| print(" $", " ".join(cmd)) | |
| subprocess.run(cmd, check=True) | |
| def install(entry: dict, force: bool = False) -> list[str]: | |
| """Fetch whatever verify() reports missing/mismatched (or everything with force=True), then re-verify.""" | |
| kind, src = entry["kind"], entry.get("source", {}) | |
| if kind == "pip": | |
| if force or verify(entry): | |
| _pip([*src.get("pip_args", []), f"{entry['name']}=={entry['version']}"]) | |
| elif kind == "git": | |
| if force or verify(entry): | |
| _pip([*src.get("pip_args", []), f"git+{entry['url']}@{entry['version']}"]) | |
| else: | |
| d = file_dir(entry) | |
| d.mkdir(parents=True, exist_ok=True) | |
| want = [f for f in file_records(entry) if force or not (d / f["name"]).exists() or sha256_file(d / f["name"]) != f["sha256"]] | |
| if want and kind == "gdown" and src.get("folder_id"): | |
| import gdown # noqa: WPS433 (optional dependency, only needed for Drive folders) | |
| with tempfile.TemporaryDirectory() as td: | |
| gdown.download_folder(id=src["folder_id"], output=td, quiet=False, use_cookies=False) | |
| for f in want: | |
| hits = list(Path(td).rglob(f["name"])) | |
| if hits: | |
| shutil.copyfile(hits[0], d / f["name"]) | |
| if want and kind == "hf": | |
| from huggingface_hub import hf_hub_download | |
| with tempfile.TemporaryDirectory() as td: | |
| for f in want: | |
| p = hf_hub_download(src["repo_id"], f.get("hf_filename", f["name"]), revision=src.get("revision"), cache_dir=td) | |
| shutil.copyfile(p, d / f["name"]) | |
| for f in want: # plain http(s) fall-back for anything still missing that has a url | |
| p = d / f["name"] | |
| if (not p.exists() or sha256_file(p) != f["sha256"]) and f.get("url", "").startswith("http"): | |
| print(f" downloading {f['url']}") | |
| _download(f["url"], p) | |
| if kind == "url" and not want and force is False: | |
| pass | |
| problems = verify(entry) | |
| if not problems: | |
| entry["installed_utc"] = entry.get("installed_utc") or utc_now() | |
| return problems | |
| # --------------------------------------------------------------------------------------------------------------- CLI | |
| def _fmt(entry: dict, problems: list[str] | None = None) -> str: | |
| sha = (entry.get("sha256") or "")[:12] | |
| st = "" if problems is None else (" OK" if not problems else " FAIL: " + "; ".join(problems)) | |
| return f"{entry['id']:<48} {entry['kind']:<5} {str(entry.get('version') or '')[:24]:<24} {sha:<12} {entry.get('license','')[:28]:<28}{st}" | |
| def main(argv=None) -> int: | |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) | |
| ap.add_argument("--list", action="store_true") | |
| ap.add_argument("--verify", action="store_true") | |
| ap.add_argument("--install", metavar="ID|all") | |
| ap.add_argument("--force", action="store_true", help="with --install: re-download / re-install even if verified") | |
| ap.add_argument("--json", action="store_true", help="print the summary as JSON") | |
| a = ap.parse_args(argv) | |
| reg = load() | |
| es = reg["entries"] | |
| if a.install: | |
| sel = es if a.install == "all" else [get(a.install)] | |
| rc = 0 | |
| for e in sel: | |
| print(f"[install] {e['id']}") | |
| probs = install(e, force=a.force) | |
| print(_fmt(e, probs)) | |
| rc |= bool(probs) | |
| save(reg) | |
| return rc | |
| if a.verify: | |
| rows = [(e, verify(e)) for e in es] | |
| if a.json: | |
| print(json.dumps({e["id"]: (p or "OK") for e, p in rows}, indent=1)) | |
| else: | |
| for e, p in rows: | |
| print(_fmt(e, p)) | |
| n_bad = sum(bool(p) for _, p in rows) | |
| print(f"{len(rows) - n_bad}/{len(rows)} entries verified OK") | |
| return int(n_bad > 0) | |
| # default: --list | |
| if a.json: | |
| print(json.dumps([{k: e.get(k) for k in FIELDS} for e in es], indent=1, ensure_ascii=False)) | |
| else: | |
| print(f"{'id':<48} {'kind':<5} {'version':<24} {'sha256[:12]':<12} license") | |
| for e in es: | |
| print(_fmt(e)) | |
| print(f"{len(es)} entries; registry {REGISTRY_JSON}") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 11.7 kB
- Xet hash:
- c9e81f1f2d9533ecf440655b0f4c3a96fbf21be24e946699957ab3e4e7b3a1ad
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.