File size: 5,074 Bytes
58ba6df
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
# SPDX-License-Identifier: Apache-2.0
"""Accuracy sweep on the REAL checkpoint over a wider input set than bench.py's 4 crops.

Inputs: the 8 annotated hands of media/bboxes.json (both hands per image, left hands mirrored
as in serving) x 4 bbox variants (as annotated, shifted +10% x, shifted -10% y, scaled 1.15)
= 32 crops.  Every crop goes through the served TtHamer path (trace replay, fresh input each
call) and is compared to the torch fp32 CPU reference: PCC of the flat 157-d output, plus the
max abs error of the rotation matrices / betas / camera.  The CPU reference outputs are cached
in --ref-cache (keyed on the checkpoint path; delete to recompute) because they take ~2 min.

Prints per-crop PCC and a summary line ``ACC {json}`` (mean / min / count below 0.9999).
Model knobs come from the environment (TT_FUSED_*), exactly like serving.
"""
from __future__ import annotations

import argparse
import json
import os
import statistics
import sys
from pathlib import Path

_REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(_REPO))


def crops():
    from hamer.preprocess import preprocess
    media = _REPO.parent / "media"
    boxes = json.loads((media / "bboxes.json").read_text())
    out = []
    for name in sorted(boxes):
        for hi, e in enumerate(boxes[name]):
            x1, y1, x2, y2 = e["bbox"]
            w, h = x2 - x1, y2 - y1
            cx, cy = (x1 + x2) / 2, (y1 + y2) / 2
            variants = {
                "orig": (x1, y1, x2, y2),
                "dx+10": (x1 + 0.1 * w, y1, x2 + 0.1 * w, y2),
                "dy-10": (x1, y1 - 0.1 * h, x2, y2 - 0.1 * h),
                "s1.15": (cx - 0.575 * w, cy - 0.575 * h, cx + 0.575 * w, cy + 0.575 * h),
            }
            for vn, bb in variants.items():
                x, _ = preprocess(media / name, bbox_xyxy=bb, is_right=bool(e.get("is_right", 1)))
                out.append((f"{name}#{hi}:{vn}", x))
    return out


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--ref-cache", default=str(Path(os.environ.get("TMPDIR", "/tmp")) / "hamer_acc_ref.pt"))
    ap.add_argument("--json", default=None)
    a = ap.parse_args()
    os.environ.setdefault("TT_FUSED_HASH", "0")
    import torch
    from hamer import Hamer, build_from_reference, load_hamer_checkpoint

    hf = Path.home() / ".cache/huggingface/hub/models--changh95--hamer-weights/snapshots/e8217b46d54b4e8afaaf0dd6efb6b62341c97d32"
    torch.manual_seed(0)
    ref = Hamer().eval()
    load_hamer_checkpoint(ref, hf / "hamer.ckpt", hf / "mano_mean_params.npz")
    xs = crops()
    cache = Path(a.ref_cache)
    refs = None
    if cache.exists():
        d = torch.load(cache)
        if isinstance(d, dict) and d.get("names") == [n for n, _ in xs] and d.get("ckpt") == str(hf):
            refs = d["out"]
    if refs is None:
        with torch.no_grad():
            refs = [ref(x) for _, x in xs]
        torch.save({"names": [n for n, _ in xs], "ckpt": str(hf), "out": refs}, cache)

    def pcc(u, v):
        u = u.reshape(-1).double(); v = v.reshape(-1).double()
        u = u - u.mean(); v = v - v.mean()
        return float((u @ v) / (u.norm() * v.norm()).clamp_min(1e-12))

    tt = build_from_reference(ref, device_id=0, strict=True)
    print(f"[acc] {tt.fused_opts.describe()}")
    res = {}
    with torch.no_grad():
        tt(xs[0][1]); tt(xs[1][1])  # JIT + trace capture
        assert tt.trace_active
        for (n, x), r in zip(xs, refs):
            o = tt(x)
            d = (o - r).abs().reshape(-1)
            res[n] = {"pcc": pcc(o, r), "rot_max": float(d[:144].max()), "betas_max": float(d[144:154].max()),
                      "cam_max": float(d[154:].max()),
                      "cam_signed": [float(v) for v in (o - r).reshape(-1)[154:157]],
                      "betas_signed": [float(v) for v in (o - r).reshape(-1)[144:154]]}
            print(f"{n:28s} pcc={res[n]['pcc']:.6f} rot={res[n]['rot_max']:.4f} betas={res[n]['betas_max']:.4f} cam={res[n]['cam_max']:.4f}")
    tt.close()
    p = [v["pcc"] for v in res.values()]
    summ = {"n": len(p), "mean": statistics.mean(p), "min": min(p), "below_0.9999": sum(v < 0.9999 for v in p),
            "rot_max": max(v["rot_max"] for v in res.values()), "cam_max": max(v["cam_max"] for v in res.values()),
            "betas_max": max(v["betas_max"] for v in res.values())}
    # signed error of the weak-perspective camera (scale s ~ 5-6, tx, ty): a consistent sign = a systematic bias
    cs = [v["cam_signed"] for v in res.values()]
    summ["cam_s_mean"] = statistics.mean(c[0] for c in cs)
    summ["cam_s_absmean"] = statistics.mean(abs(c[0]) for c in cs)
    summ["cam_s_pos"] = sum(c[0] > 0 for c in cs)
    summ["cam_t_absmean"] = statistics.mean((abs(c[1]) + abs(c[2])) / 2 for c in cs)
    summ["betas_signed_mean"] = statistics.mean(statistics.mean(v["betas_signed"]) for v in res.values())
    print("ACC " + json.dumps(summ))
    if a.json:
        Path(a.json).write_text(json.dumps({"summary": summ, "per_crop": res}, indent=1))


if __name__ == "__main__":
    main()