File size: 4,870 Bytes
d456972 | 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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | #!/usr/bin/env python3
"""Download and verify the published MotionBricks GGUF/style distribution."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import shutil
import sys
import urllib.parse
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
DEFAULT_REPO = "LocalAI-io/MotionBricks-G1-GGML"
DEFAULT_REVISION = "cc2a47603dbc203a4f18f35dd06ed3611833f506"
DEFAULT_MANIFEST = ROOT / "scripts/hf/MotionBricks-G1-GGML/MANIFEST.json"
FORMAT = "motionbricks-gguf-distribution-v1"
UPSTREAM_REVISION = "a0732b642c0333077e127a2f56ab0014c196bca4"
def digest(path: Path) -> str:
value = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(8 << 20), b""):
value.update(block)
return value.hexdigest()
def valid(path: Path, entry: dict[str, object]) -> bool:
return (path.is_file() and path.stat().st_size == entry["bytes"]
and digest(path) == entry["sha256"])
def safe_relative(value: object) -> Path:
path = Path(str(value))
if path.is_absolute() or ".." in path.parts or not path.parts:
raise ValueError(f"unsafe distribution path: {value!r}")
if path.parts[0] not in {"g1-f32", "styles"}:
raise ValueError(f"unexpected distribution path: {value!r}")
if path.suffix not in {".gguf", ".mbstyle", ".json"}:
raise ValueError(f"unexpected distribution file type: {value!r}")
return path
def load_manifest(path: Path, repo: str) -> dict[str, object]:
value = json.loads(path.read_text(encoding="utf-8"))
if value.get("format") != FORMAT:
raise ValueError(f"unsupported manifest format in {path}")
if value.get("repository") != repo:
raise ValueError(f"manifest is for {value.get('repository')}, not {repo}")
if value.get("upstream_revision") != UPSTREAM_REVISION:
raise ValueError("manifest refers to an unexpected upstream revision")
files = value.get("files")
if not isinstance(files, list) or not files:
raise ValueError("manifest has no files")
for entry in files:
if (not isinstance(entry, dict) or not isinstance(entry.get("bytes"), int)
or entry["bytes"] <= 0 or not isinstance(entry.get("sha256"), str)
or len(entry["sha256"]) != 64):
raise ValueError("manifest contains a malformed file record")
safe_relative(entry.get("path"))
return value
def download(url: str, destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
temporary = destination.with_name(destination.name + ".download")
request = urllib.request.Request(url, headers={"User-Agent": "motion-bricks.cpp/0.1"})
try:
with urllib.request.urlopen(request) as response, temporary.open("wb") as output:
shutil.copyfileobj(response, output, length=8 << 20)
output.flush()
os.fsync(output.fileno())
os.replace(temporary, destination)
finally:
temporary.unlink(missing_ok=True)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output", type=Path, default=ROOT / "generated",
help="destination containing g1-f32/ and styles/")
parser.add_argument("--repo", default=DEFAULT_REPO, help="Hugging Face model repository")
parser.add_argument("--revision", default=DEFAULT_REVISION,
help="pinned Hugging Face commit or revision")
parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST,
help="version-controlled distribution manifest")
args = parser.parse_args()
try:
manifest = load_manifest(args.manifest, args.repo)
quoted_repo = urllib.parse.quote(args.repo, safe="/")
quoted_revision = urllib.parse.quote(args.revision, safe="")
base = f"https://huggingface.co/{quoted_repo}/resolve/{quoted_revision}"
for entry in manifest["files"]:
relative = safe_relative(entry["path"])
destination = args.output / relative
if valid(destination, entry):
print(f"verified {destination}")
continue
print(f"downloading {relative} from {args.repo}@{args.revision}")
download(f"{base}/{urllib.parse.quote(relative.as_posix(), safe='/')}", destination)
if not valid(destination, entry):
raise ValueError(f"downloaded file failed size/SHA-256 verification: {destination}")
except (OSError, ValueError, urllib.error.URLError, json.JSONDecodeError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
print("MotionBricks GGUF and style bundles are present and verified")
return 0
if __name__ == "__main__":
sys.exit(main())
|