scripts/fetch_hf_model.py

#!/usr/bin/env python3
"""Download and verify the Hugging Face crowd-perception model.

Run this once, with network access, before the demo. It walks the candidate
chain in `flowtwin/perception/huggingface.py`, loads the first model that
works, runs one real inference to prove the whole path end to end, and writes
`models/perception_manifest.json` recording which model was selected and why.

If every candidate fails it says so plainly and prints each error. FlowTwin
then reports perception as unavailable at runtime rather than inventing a count.

Run:  python scripts/fetch_hf_model.py [--model REPO_ID]
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "backend"))

from flowtwin.config import PERCEPTION_SAMPLE_DIR, SETTINGS          # noqa: E402
from flowtwin.perception.huggingface import (                        # noqa: E402
    CANDIDATES,
    MANIFEST_PATH,
    CrowdPerception,
)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--model", default=None, help="force a specific repo id")
    ap.add_argument("--sample", default=None, help="image to test with")
    args = ap.parse_args()

    cfg = SETTINGS.perception
    if args.model:
        cfg = type(cfg)(enabled=True, override_model=args.model,
                        cache_dir=cfg.cache_dir,
                        max_image_pixels=cfg.max_image_pixels)

    print("Candidate chain (first one that loads wins):")
    for c in CANDIDATES:
        print(f"  · {c.repo_id}\n      {c.label} — {c.note}")
    print()

    perception = CrowdPerception(cfg)
    perception._ensure_loaded()
    status = perception.status()

    if not status["loaded"]:
        print("No model could be loaded.\n")
        for attempt in status["attempts"]:
            print(f"  ✗ {attempt['repo_id']}\n      {attempt['error']}")
        print("\nCommon causes: no network access to huggingface.co, `torch` or "
              "`transformers` not installed, or a private/renamed repository.")
        print("FlowTwin will run normally; the perception panel will report "
              "itself unavailable rather than showing a fabricated count.")
        return 1

    print(f"Loaded: {status['model']}\n  {status['label']}\n  {status['note']}\n")

    sample_path = Path(args.sample) if args.sample else None
    if sample_path is None:
        candidates = (sorted(PERCEPTION_SAMPLE_DIR.glob("*.jpg"))
                      + sorted(PERCEPTION_SAMPLE_DIR.glob("*.png")))
        sample_path = candidates[0] if candidates else None

    if sample_path is None or not sample_path.exists():
        print("No sample image available to verify inference. Drop a crowd photo "
              f"into {PERCEPTION_SAMPLE_DIR} and re-run, or upload one from the "
              "dashboard's perception panel.")
        return 0

    print(f"Verifying inference on {sample_path.name} …")
    result = perception.analyze(sample_path.read_bytes(), None, None, None, sample_path.name)
    if not result.get("ok"):
        print(f"  ✗ inference failed: {result.get('error')}")
        return 1
    obs = result["observation"]
    print(f"  ✓ counted {obs['people']} people in {result['latency_ms']:.0f} ms "
          f"({result['detail'].get('method')})")
    print(f"\nManifest written to {MANIFEST_PATH}")
    print(json.dumps(json.loads(MANIFEST_PATH.read_text()), indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())