AsianPlayer's picture
download
raw
22.3 kB
"""MotionBERT (released weights) on the AIDE keypoint protocol — faithful-baseline adapter.
Pre-registration: docs/decisions.md 2026-09-16 (2026-09-16T21:28:48Z, 'faithful MotionBERT baseline'). Owner's rule: the released weights must
be used and fine-tuned on our data before a comparison counts as a reproduction. The existing `aide/models/motionbert_kp.py` (d 128, 3 blocks,
no released weights) stays what it is: an architecture-FAMILY control. This module is the reproduction:
* backbone = the official DSTformer (vendored verbatim: `DSTformer.py`, Apache-2.0, see NOTICE) with the released configuration
configs/pretrain/MB_pretrain.yaml (dim_feat 512, depth 5, num_heads 8, mlp_ratio 2, maxlen 243, dim_rep 512, att_fuse) and the
LayerNorm eps 1e-6 of lib/utils/learning.load_backbone;
* weights = checkpoint/pretrain/MB_release/latest_epoch.bin (Hugging Face mirror of the README's OneDrive release; URL / sha256 / size in
CKPTS and NOTICE). `load_released` loads every tensor of the checkpoint by name with a strict report: the 240 block tensors, the
temporal position embedding, joints_embed, norm, pre_logits and the 5 ts_attn fusers must load exactly (asserted); the 17-row
spatial position embedding is loaded exactly and then EXPANDED to 136 rows by copying (see joint map); the pretrained 3-D lifting
head (Linear 512 -> 3) is dropped and replaced by the new 2-D read-out. Nothing else may be new.
* input = AIDE HALPE-136 (x, y px, conf) of the T_OBS = 5 observed frames -> MotionBERT's in-the-wild `--pixel` normalisation
(lib/data/dataset_wild.read_input with vid_size = (W, H): xy' = (xy - (W, H)/2) / (min(W, H)/2), i.e. y in [-1, 1] over the image
height, aspect ratio preserved; confidence kept as channel 3, exactly as the released model was fed (dim_in 3)). The repo's other
normalisation, crop_scale (square crop of the clip's keypoint bounding box), is NOT used because it is not a fixed map (its
inverse depends on the clip) — the pixel map is a fixed affine map and its inverse gives the absolute read-out its pixels.
* joints = the 136 HALPE keypoints stay 136 tokens (HALPE order). Each token receives the pretrained spatial position embedding of ONE
H36M-17 joint: the body joints through the repo's own halpe2h36m (lib/data/dataset_wild.py; HALPE-26 is the first 26 joints of
HALPE-136), the other keypoints from the nearest pretrained body joint within the pre-registered candidate set (face <- nose /
head, eyes-ears <- nose / head, hands <- wrists, feet <- ankles; head-top / neck / pelvis are direct halpe2h36m rows), ties by the
TRAIN-split mean pixel distance — `joint_map.json` next to this file records every row (`python adapter.py --write-map`).
* time = the 5 observed frames + 5 placeholder frames that are copies of the LAST observed frame (zero-velocity initial guess) -> 10 frames
-> temp_embed[:10] (pretrained rows). Only the observed frames ever enter the model (forward asserts T_OBS frames).
* read-out = per-frame 512-d representation of the 5 placeholder positions (after the released pre_logits fc + tanh, as
DSTformer.get_representation) -> Linear(512 -> 2):
(a) formulation="residual": pred = last observed xy + SCALE * head(rep), head ZERO-initialised (exactly zero-velocity at init);
(b) formulation="abs": pred = denormalise(head(rep)) px, head at PyTorch default init (their formulation: absolute
coordinates in the normalised frame, no residual, no zero-velocity start);
(c) init="random": same architecture, official random init (_init_weights, trunc_normal 0.02), residual read-out.
Tensor output -> `train_kp`'s plain Euclidean loss; the model exposes `param_groups()` (backbone / head) for the two learning rates.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import sys
import time
from functools import partial
from pathlib import Path
import numpy as np
import torch
import torch.nn as nn
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
from aide.data import protocol as P # noqa: E402
from aide.models.motionbert_official.DSTformer import DSTformer # noqa: E402
HERE = Path(__file__).resolve().parent
JOINT_MAP_JSON = HERE / "joint_map.json"
CKPT_DIR = P.AIDE_PKG / "checkpoints" / "motionbert"
HF = "https://huggingface.co/walterzhu/MotionBERT/resolve/main/"
CKPTS = {
"MB_release": {"path": CKPT_DIR / "checkpoint/pretrain/MB_release/latest_epoch.bin", "url": HF + "checkpoint/pretrain/MB_release/latest_epoch.bin",
"sha256": "02a9ee017e5c9d688b5dd41e0055bddb75b6eee7f5eed326a8cdc6e5bf08eaf8", "bytes": 169958969,
"note": "released pretraining checkpoint (README: MotionBERT (162MB), configs/pretrain/MB_pretrain.yaml)"},
"MB_ft_h36m": {"path": CKPT_DIR / "checkpoint/pose3d/FT_MB_release_MB_ft_h36m/best_epoch.bin", "url": HF + "checkpoint/pose3d/FT_MB_release_MB_ft_h36m/best_epoch.bin",
"sha256": "597c708bd585646cc6961df057ef73634edb83a4d5c1155d98a4cc0c04e81b98", "bytes": 169958969,
"note": "fallback only (3D pose H36M-SH fine-tuned); not used while MB_release loads"},
}
# configs/pretrain/MB_pretrain.yaml + lib/utils/learning.load_backbone (dim_in 3, dim_out 3, norm eps 1e-6, num_joints 17)
CFG = dict(dim_in=3, dim_out=3, dim_feat=512, dim_rep=512, depth=5, num_heads=8, mlp_ratio=2, num_joints=17, maxlen=243, att_fuse=True)
SCALE = 200.0 # px per unit of the residual read-out (as every shared-recipe baseline)
CENTRE = (P.W / 2.0, P.H / 2.0)
PIX_SCALE = min(P.W, P.H) / 2.0 # 540 px: lib/data/dataset_wild.read_input(vid_size) / infer_wild.py --pixel
H36M_NAMES = ("Hip", "RHip", "RKnee", "RAnkle", "LHip", "LKnee", "LAnkle", "Belly", "Neck", "Nose", "Head",
"LShoulder", "LElbow", "LWrist", "RShoulder", "RElbow", "RWrist")
HALPE_BODY_NAMES = ("Nose", "LEye", "REye", "LEar", "REar", "LShoulder", "RShoulder", "LElbow", "RElbow", "LWrist", "RWrist", "LHip", "RHip",
"LKnee", "RKnee", "LAnkle", "RAnkle", "Head", "Neck", "Hip", "LBigToe", "RBigToe", "LSmallToe", "RSmallToe", "LHeel", "RHeel")
# lib/data/dataset_wild.halpe2h36m, H36M index -> HALPE source (H36M 7 'Belly' = 0.5 * (Neck 18 + Hip 19) has no HALPE joint)
H36M_FROM_HALPE = {0: 19, 1: 12, 2: 14, 3: 16, 4: 11, 5: 13, 6: 15, 8: 18, 9: 0, 10: 17, 11: 5, 12: 7, 13: 9, 14: 6, 15: 8, 16: 10}
HALPE2H36M_DIRECT = {h: m for m, h in H36M_FROM_HALPE.items()} # 16 HALPE body joints with a pretrained row of their own
# pre-registered candidate sets for the 120 remaining keypoints (nearest pretrained body joint; the TRAIN-split mean distance decides in a set)
CANDIDATES = {
"eyes_ears": ([1, 2, 3, 4], [9, 10]),
"feet_left": ([20, 22, 24], [6]),
"feet_right": ([21, 23, 25], [3]),
"face": (list(range(26, 94)), [9, 10]),
"lhand": (list(range(94, 115)), [13]),
"rhand": (list(range(115, 136)), [16]),
}
DEFAULT_IN_SET = {"eyes_ears": 9, "face": 9, "feet_left": 6, "feet_right": 3, "lhand": 13, "rhand": 16} # used only if joint_map.json is absent
# ----------------------------------------------------------------------------- normalisation (repo --pixel map and its inverse)
def normalise(obs: torch.Tensor) -> torch.Tensor:
"""[..., 136, 3] (x px, y px, conf) -> MotionBERT --pixel frame: xy' = (xy - (W, H)/2) / (min(W, H)/2); conf unchanged."""
c = obs.new_tensor(CENTRE)
return torch.cat([(obs[..., :2] - c) / PIX_SCALE, obs[..., 2:3]], -1)
def denormalise_xy(xy: torch.Tensor) -> torch.Tensor:
"""inverse of `normalise` for the absolute read-out: px = xy' * min(W, H)/2 + (W, H)/2 (infer_wild.py --pixel back-projection)."""
return xy * PIX_SCALE + xy.new_tensor(CENTRE)
def halpe2h36m_np(x: np.ndarray) -> np.ndarray:
"""lib/data/dataset_wild.halpe2h36m for [..., >=26, C] arrays (used to build / verify the joint map)."""
y = np.zeros(x.shape[:-2] + (17, x.shape[-1]), dtype=x.dtype)
for m, h in H36M_FROM_HALPE.items():
y[..., m, :] = x[..., h, :]
y[..., 7, :] = 0.5 * (x[..., 18, :] + x[..., 19, :])
return y
# ----------------------------------------------------------------------------- joint map (136 HALPE tokens -> 17 pretrained rows)
def default_joint_map() -> np.ndarray:
jm = np.full(P.J, -1, dtype=np.int64)
for h, m in HALPE2H36M_DIRECT.items():
jm[h] = m
for grp, (idx, _) in CANDIDATES.items():
jm[idx] = DEFAULT_IN_SET[grp]
assert (jm >= 0).all()
return jm
def load_joint_map(path: Path = JOINT_MAP_JSON) -> np.ndarray:
if not Path(path).exists():
return default_joint_map()
d = json.loads(Path(path).read_text())
jm = np.array([d["rows"][str(j)]["h36m"] for j in range(P.J)], dtype=np.int64)
assert jm.shape == (P.J,) and jm.min() >= 0 and jm.max() < 17
return jm
def build_joint_map(S_train: np.ndarray) -> dict:
"""TRAIN-split geometry: mean pixel distance of every HALPE keypoint to each H36M-17 joint (halpe2h36m of the same frame); nearest overall
(diagnostic) and nearest within the pre-registered candidate set (the row used). S_train: [N, T, 136, 3] px."""
x = S_train[..., :2].astype(np.float64).reshape(-1, P.J, 2) # [NT, 136, 2]
h = halpe2h36m_np(S_train[..., :2].astype(np.float64)).reshape(-1, 17, 2) # [NT, 17, 2]
dist = np.sqrt(((x[:, :, None, :] - h[:, None, :, :]) ** 2).sum(-1)).mean(0) # [136, 17]
rows = {}
for j in range(P.J):
if j in HALPE2H36M_DIRECT:
m = HALPE2H36M_DIRECT[j]
rows[str(j)] = {"halpe": HALPE_BODY_NAMES[j], "h36m": m, "h36m_name": H36M_NAMES[m], "rule": "direct: lib/data/dataset_wild.halpe2h36m",
"candidates": [m], "mean_dist_px": {H36M_NAMES[m]: round(float(dist[j, m]), 2)}, "nearest_overall": H36M_NAMES[int(dist[j].argmin())]}
for grp, (idx, cand) in CANDIDATES.items():
for j in idx:
m = int(cand[int(np.argmin([dist[j, c] for c in cand]))])
nm = HALPE_BODY_NAMES[j] if j < 26 else (f"face_{j - 26}" if j < 94 else (f"lhand_{j - 94}" if j < 115 else f"rhand_{j - 115}"))
rows[str(j)] = {"halpe": nm, "h36m": m, "h36m_name": H36M_NAMES[m], "group": grp,
"rule": f"nearest pretrained body joint within {{{', '.join(H36M_NAMES[c] for c in cand)}}} by TRAIN mean distance",
"candidates": list(map(int, cand)), "mean_dist_px": {H36M_NAMES[c]: round(float(dist[j, c]), 2) for c in cand},
"nearest_overall": H36M_NAMES[int(dist[j].argmin())]}
assert len(rows) == P.J
counts = {H36M_NAMES[m]: int(sum(1 for r in rows.values() if r["h36m"] == m)) for m in range(17)}
return {"generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "source": "aide/models/motionbert_official/adapter.py --write-map",
"n_train_frames": int(x.shape[0]), "layout": "HALPE-136: 0-16 COCO body, 17 head-top, 18 neck, 19 pelvis, 20-25 feet, 26-93 face (iBUG-68), 94-114 left hand, 115-135 right hand",
"h36m_names": list(H36M_NAMES), "halpe2h36m": {H36M_NAMES[m]: (HALPE_BODY_NAMES[h] if h is not None else "0.5*(Neck+Hip)") for m, h in
[(m, H36M_FROM_HALPE.get(m)) for m in range(17)]},
"tokens_per_pretrained_row": counts, "unused_pretrained_rows": [H36M_NAMES[m] for m in range(17) if counts[H36M_NAMES[m]] == 0], "rows": rows}
def expand_pos_embed(backbone: DSTformer, joint_map: np.ndarray) -> None:
"""[1, 17, d] pretrained (or random) spatial embedding -> [1, 136, d]: row j := row joint_map[j] (copies; every row trainable afterwards)."""
pe = backbone.pos_embed.data
assert tuple(pe.shape) == (1, CFG["num_joints"], CFG["dim_feat"]), tuple(pe.shape)
idx = torch.as_tensor(joint_map, dtype=torch.long, device=pe.device)
backbone.pos_embed = nn.Parameter(pe[:, idx].clone())
# ----------------------------------------------------------------------------- weights
def sha256_file(p: Path) -> str:
h = hashlib.sha256()
with open(p, "rb") as f:
for b in iter(lambda: f.read(1 << 22), b""):
h.update(b)
return h.hexdigest()
def tensor_sha256(t: torch.Tensor) -> str:
return hashlib.sha256(t.detach().cpu().contiguous().float().numpy().tobytes()).hexdigest()
def read_checkpoint(which: str = "MB_release") -> tuple[dict, dict]:
meta = CKPTS[which]
path = Path(meta["path"])
if not path.exists():
raise FileNotFoundError(f"released MotionBERT checkpoint missing: download {meta['url']} (sha256 {meta['sha256']}, {meta['bytes']} bytes) to {path}")
try:
ck = torch.load(path, map_location="cpu", weights_only=True)
except Exception: # older pickles: fall back (the file's sha256 is verified against the published LFS digest)
ck = torch.load(path, map_location="cpu", weights_only=False)
sd = ck["model_pos"] if isinstance(ck, dict) and "model_pos" in ck else (ck["model"] if isinstance(ck, dict) and "model" in ck else ck)
sd = {(k[len("module."):] if k.startswith("module.") else k): v for k, v in sd.items()}
return sd, meta
def load_released(backbone: DSTformer, which: str = "MB_release", verify_sha: bool = True) -> dict:
"""Strict load of the released checkpoint into a fresh 17-joint DSTformer (call BEFORE expand_pos_embed). Returns the report."""
sd, meta = read_checkpoint(which)
file_sha = sha256_file(Path(meta["path"]))
size = Path(meta["path"]).stat().st_size
if verify_sha:
assert file_sha == meta["sha256"], (file_sha, meta["sha256"])
assert size == meta["bytes"], (size, meta["bytes"])
own = backbone.state_dict()
to_load, dropped, mismatched = {}, [], []
for k, v in sd.items():
if k in ("head.weight", "head.bias"):
dropped.append(k) # the pretrained 3-D lifting head (512 -> 3) is replaced by the new 2-D read-out
continue
if k not in own:
dropped.append(k)
continue
if tuple(own[k].shape) != tuple(v.shape):
mismatched.append(k)
continue
to_load[k] = v
missing = [k for k in own if k not in to_load and not k.startswith("head.")]
assert not mismatched, f"shape mismatch: {mismatched}"
assert not missing, f"backbone tensors absent from the checkpoint: {missing}"
assert sorted(dropped) == ["head.bias", "head.weight"], f"unexpected dropped keys: {dropped}"
res = backbone.load_state_dict(to_load, strict=False)
assert sorted(res.missing_keys) == ["head.bias", "head.weight"] and not res.unexpected_keys, (res.missing_keys, res.unexpected_keys)
n_blocks = sum(1 for k in to_load if k.startswith(("blocks_st.", "blocks_ts.")))
assert n_blocks == 240 and "temp_embed" in to_load and "pos_embed" in to_load, n_blocks
for pref in ("joints_embed.", "norm.", "pre_logits.fc.", "ts_attn."):
assert any(k.startswith(pref) for k in to_load), pref
# post-load identity check on every loaded tensor (bit-exact)
own = backbone.state_dict()
for k, v in to_load.items():
assert torch.equal(own[k], v), k
probe = "blocks_st.0.attn_s.qkv.weight"
return {"checkpoint": which, "path": str(meta["path"]), "url": meta["url"], "sha256": file_sha, "sha256_expected": meta["sha256"], "bytes": size,
"n_ckpt_tensors": len(sd), "n_loaded_exact": len(to_load), "n_params_loaded": int(sum(v.numel() for v in to_load.values())),
"loaded_groups": {"blocks_st+blocks_ts": n_blocks, "temp_embed": 1, "pos_embed (17 rows, expanded afterwards)": 1,
"joints_embed": sum(k.startswith("joints_embed.") for k in to_load), "norm": sum(k.startswith("norm.") for k in to_load),
"pre_logits": sum(k.startswith("pre_logits.") for k in to_load), "ts_attn": sum(k.startswith("ts_attn.") for k in to_load)},
"dropped": sorted(dropped), "shape_mismatch": mismatched, "missing_in_ckpt": missing, "probe_tensor": probe, "probe_sha256": tensor_sha256(to_load[probe])}
# ----------------------------------------------------------------------------- model
class MotionBERTOfficial(nn.Module):
"""Official DSTformer + released weights on 136 HALPE tokens x (5 observed + 5 placeholder) frames; read-out on the 5 placeholders."""
name = "motionbert_official"
def __init__(self, formulation: str = "residual", init: str = "released", ckpt: str = "MB_release", joint_map: np.ndarray | None = None,
T: int = P.T_OBS, H: int = P.T_PRED, J: int = P.J, verify_sha: bool = True):
super().__init__()
assert formulation in ("residual", "abs"), formulation
assert init in ("released", "random"), init
self.T, self.H, self.J, self.formulation, self.init_mode = T, H, J, formulation, init
self.backbone = DSTformer(norm_layer=partial(nn.LayerNorm, eps=1e-6), **CFG) # official init (_init_weights) until loaded
self.load_report = load_released(self.backbone, ckpt, verify_sha) if init == "released" else None
jm = load_joint_map() if joint_map is None else np.asarray(joint_map, dtype=np.int64)
assert jm.shape == (J,), jm.shape
self.register_buffer("joint_map", torch.as_tensor(jm), persistent=False)
expand_pos_embed(self.backbone, jm)
self.backbone.head = nn.Identity() # pretrained 3-D lifting head dropped
self.head = nn.Linear(CFG["dim_rep"], 2) # new 2-D read-out
if formulation == "residual":
nn.init.zeros_(self.head.weight)
nn.init.zeros_(self.head.bias) # exactly zero-velocity at init
self.new_parameters = ["head.weight", "head.bias"]
self.copied_parameters = ["backbone.pos_embed (17 pretrained rows -> 136 rows by copying joint_map)"]
# two learning rates (backbone 1e-4 / head 1e-3 pre-registered)
def param_groups(self):
return {"backbone": list(self.backbone.parameters()), "head": list(self.head.parameters())}
def build_input(self, obs: torch.Tensor) -> torch.Tensor:
"""obs [B, T_OBS, 136, 3] px -> [B, T_OBS + T_PRED, 136, 3] normalised; frames >= T_OBS are copies of the last observed frame."""
assert obs.dim() == 4 and obs.shape[1] == self.T and obs.shape[2] == self.J, tuple(obs.shape)
x = normalise(obs)
return torch.cat([x, x[:, -1:].expand(-1, self.H, -1, -1)], 1)
def forward(self, obs: torch.Tensor) -> torch.Tensor:
last = obs[:, -1:, :, :2]
x = self.build_input(obs)
rep = self.backbone(x, return_rep=True) # [B, 10, 136, 512] (pre_logits fc + tanh)
y = self.head(rep[:, self.T:]) # [B, 5, 136, 2] read-out on the placeholder positions
if self.formulation == "abs":
return denormalise_xy(y)
return last + SCALE * y
def build(formulation: str = "residual", init: str = "released", **kw) -> MotionBERTOfficial:
return MotionBERTOfficial(formulation=formulation, init=init, **kw)
def weight_summary(model: MotionBERTOfficial) -> dict:
"""Run-json block: what was loaded / copied / new (counts from the model itself)."""
n_total = sum(p.numel() for p in model.parameters())
n_head = sum(p.numel() for p in model.head.parameters())
n_pos = model.backbone.pos_embed.numel()
rep = dict(model.load_report or {"checkpoint": None, "note": "random init (official _init_weights)"})
rep.update({"n_params_total": int(n_total), "n_params_new_head": int(n_head), "n_params_pos_embed_136": int(n_pos),
"n_params_pretrained_in_model": int(n_total - n_head - n_pos + (17 * CFG["dim_feat"] if model.load_report else 0)) if model.load_report else 0,
"new_parameters": model.new_parameters, "copied_parameters": model.copied_parameters,
"joint_map_json": str(JOINT_MAP_JSON), "joint_map_sha256": (sha256_file(JOINT_MAP_JSON) if JOINT_MAP_JSON.exists() else None),
"formulation": model.formulation, "init": model.init_mode})
return rep
# ----------------------------------------------------------------------------- CLI
def main(argv=None):
ap = argparse.ArgumentParser()
ap.add_argument("--write-map", action="store_true", help="build joint_map.json from the TRAIN split of the protocol cache")
ap.add_argument("--init", default="released", choices=("released", "random"))
ap.add_argument("--formulation", default="residual", choices=("residual", "abs"))
a = ap.parse_args(argv)
if a.write_map:
c = np.load(P.CACHE_NPZ, allow_pickle=False)
tr = c["split"] == "train"
S = c["S"][tr][:, P.sample_idx(P.N_FRAMES_STD)] # TRAIN clips, protocol frames only
d = build_joint_map(S)
JOINT_MAP_JSON.write_text(json.dumps(d, indent=1))
print(f"wrote {JOINT_MAP_JSON}: tokens per pretrained row {d['tokens_per_pretrained_row']}; unused rows {d['unused_pretrained_rows']}")
diff = [(k, r["h36m_name"], r["nearest_overall"]) for k, r in d["rows"].items() if r["h36m_name"] != r["nearest_overall"]]
print(f"rows whose in-set choice differs from the unrestricted nearest joint: {len(diff)} -> {diff[:12]}{' ...' if len(diff) > 12 else ''}")
return
m = build(a.formulation, a.init)
x = torch.rand(2, P.T_OBS, P.J, 3) * torch.tensor([P.W, P.H, 1.0])
y = m(x)
print(json.dumps(weight_summary(m), indent=1, default=str))
print("out", tuple(y.shape), "zero-velocity at init:", torch.equal(y, x[:, -1:, :, :2].expand_as(y)))
if __name__ == "__main__":
main()

Xet Storage Details

Size:
22.3 kB
·
Xet hash:
820d6f863697760a13e86b48e66997251fcbdc026e517fd6bbeb33796bf8d07d

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.