#!/usr/bin/env python3 """DTW + FGD + MAEJ for every TriVis condition, one protocol, one unit. Protocol (fixed for all systems, so the numbers are comparable): clips NSLP-G's 200 test clip ids, matched across dumps BY NAME joints 50 (8 body == OpenPose 0-7, plus 42 hand) -- no face unit PER-CLIP SHOULDER WIDTH: undo the per-axis frame anisotropy, subtract the reference clip's median neck, divide by its median shoulder width Why re-score instead of quoting each project: every project's own DTW picks a different alignment path, and DTW minimises the cost it is given, so the path decides the score. SignDiff aligns on HANDS ONLY (42 joints) -- most favourable NSLP-G aligns on 50 joints (84 % hands) T2M-GPT aligns on 124 joints (55 % near-rigid face) -- least favourable Measured on T2M-GPT, 124 -> 50 joints alone moved hands DTW by ~8 %. Metrics: DTW hands geometric distance after the most forgiving time alignment MAEJ* paired per-joint mean absolute error, NO time alignment, x100 FGD Frechet distance between Gaussians fitted to Transformer-AE features of real vs generated poses. Distributional, never paired -- so it penalises hedging toward a mean, which DTW rewards. AE trained on GT TRAIN only. Reported against a real-vs-real floor (GT split in half), because FGD is not zero at finite sample size, and every condition is scored on the SAME clips because FGD is sample-size biased. """ import argparse import glob import json import os import numpy as np import torch from eval_fgd_maej import TFAE, features, frechet, maej, mean_cov, train_ae from dataset import dataset_vsl REPO = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..') KEEP_50_UPPER = list(range(8)) + list(range(82, 124)) GROUPS_50 = {'all': (0, 50), 'body': (0, 8), 'hands': (8, 50)} NECK, RSHO, LSHO = 1, 2, 5 def anchor_of(xy, valid, W, H): px = xy * np.array([W, H], np.float32) ok = (valid[:, NECK] > 0) & (valid[:, RSHO] > 0) & (valid[:, LSHO] > 0) if ok.sum() < 3: ok = np.ones(len(xy), bool) neck = np.median(px[ok, NECK, :], axis=0) sw = float(np.median(np.linalg.norm(px[ok, RSHO, :] - px[ok, LSHO, :], axis=-1))) return neck, max(sw, 1e-3) def to_shoulder(xy, neck, sw, W, H): return ((xy * np.array([W, H], np.float32)) - neck[None, None, :]) / sw def dtw50(pred, gt, valid): import eval_vsl saved = eval_vsl.GROUPS eval_vsl.GROUPS = GROUPS_50 try: return eval_vsl.dtw_mje(pred.astype(np.float64), gt.astype(np.float64), valid.astype(np.float64)) finally: eval_vsl.GROUPS = saved def load_dump(p): d = np.load(p, allow_pickle=True) return {str(n): q for n, q in zip(d['names'], d['poses'])} def main(): ap = argparse.ArgumentParser() ap.add_argument('--mine', nargs='+', default=['dumps_trivis']) ap.add_argument('--nslpg-root', default=os.path.join(REPO, '0.NSLP-G/sentence-level/dumps')) ap.add_argument('--data-dir', default='./dataset/VSL_upper') ap.add_argument('--frame-w', type=float, default=1176.0) ap.add_argument('--frame-h', type=float, default=1288.0) ap.add_argument('--epochs', type=int, default=25) ap.add_argument('--device', default='cuda') ap.add_argument('--out-json', default='output_vsl/trivis_metrics.json') args = ap.parse_args() W, H = args.frame_w, args.frame_h device = torch.device(args.device) # ---------------- conditions ---------------- conds = {} for d in args.mine: pref = '' if d == args.mine[0] else os.path.basename(d).replace('dumps_trivis', '') + '/' for p in sorted(glob.glob(os.path.join(d, '*.npz'))): t = os.path.splitext(os.path.basename(p))[0] if pref and t == 'gt': continue # one shared ground truth fam = ('T2M' if t.startswith('t2m') else 'COMPOSED' if t.startswith('composed') else 'GT') conds[f'{fam}:{pref}{t}'] = load_dump(p) for sub in ('pre', 'scratch', 'shape', 'signdiff'): for p in sorted(glob.glob(os.path.join(args.nslpg_root, sub, '*.npz'))): t = os.path.splitext(os.path.basename(p))[0] src = 'SIGNDIFF' if sub == 'signdiff' else 'NSLPG' conds[f'{src}:{sub}/{t}'] = load_dump(p) gt_key = next(k for k in conds if k.endswith(':gt')) gt = conds.pop(gt_key) common = set(gt) for m in conds.values(): common &= set(m) names = sorted(common) print(f'{len(conds)} conditions, {len(names)} clips shared by all (GT has {len(gt)})') for k, m in sorted(conds.items()): print(f' {k:<34} {len(m)}') if len(names) < 20: raise SystemExit('too few shared clips') # ---------------- per-clip anchors + validity from the pack ---------------- st = dataset_vsl.VSLStore(args.data_dir, 'test') by_name = {c['name']: i for i, c in enumerate(st.index)} NKF = st.layout.n_kpts anchors, valids = {}, {} for nm in names: i = by_name[nm] motion, mask = st.get(i) v50 = mask[:, ::2][:, KEEP_50_UPPER] g50 = (motion * st.std + st.mean).reshape(-1, NKF, 2)[:, KEEP_50_UPPER] anchors[nm] = anchor_of(g50, v50, W, H) valids[nm] = v50 def sh(nm, xy): neck, sw = anchors[nm] return to_shoulder(np.asarray(xy, np.float32), neck, sw, W, H) gt_sh = {nm: sh(nm, gt[nm]) for nm in names} # ---------------- FGD feature extractor: GT TRAIN, shoulder space ---------------- tr = dataset_vsl.VSLStore(args.data_dir, 'train') seqs = [] for i in range(len(tr.index)): motion, mask = tr.get(i) v50 = mask[:, ::2][:, KEEP_50_UPPER] g50 = (motion * tr.std + tr.mean).reshape(-1, NKF, 2)[:, KEEP_50_UPPER] n, s = anchor_of(g50, v50, W, H) seqs.append(to_shoulder(g50, n, s, W, H).reshape(len(g50), -1).astype(np.float32)) print(f'AE on {len(seqs)} GT train clips, dim {seqs[0].shape[-1]}') ae = train_ae(seqs, device, seqs[0].shape[-1], epochs=args.epochs) flat = lambda a: np.asarray(a, np.float32).reshape(len(a), -1) half = len(names) // 2 fa = features(ae, [flat(gt_sh[n]) for n in names[:half]], device) fb = features(ae, [flat(gt_sh[n]) for n in names[half:]], device) floor = frechet(*mean_cov(fa), *mean_cov(fb)) mu_r, sig_r = mean_cov(features(ae, [flat(gt_sh[n]) for n in names], device)) # ---------------- score ---------------- rows = [] for tag, m in sorted(conds.items()): d = {g: [] for g in GROUPS_50} mj, lr, seqs_c = [], [], [] for nm in names: p = sh(nm, m[nm]) g = gt_sh[nm] for k, v in dtw50(p, g, valids[nm]).items(): d[k].append(v) mj.append(maej(p, g)) lr.append(len(p) / len(g)) seqs_c.append(flat(p)) fz = features(ae, seqs_c, device) rows.append({'condition': tag, 'n': len(names), 'dtw_hands': float(np.mean(d['hands'])), 'dtw_body': float(np.mean(d['body'])), 'dtw_all': float(np.mean(d['all'])), 'FGD': frechet(mu_r, sig_r, *mean_cov(fz)), 'MAEJ_x100': float(np.mean(mj)) * 100, 'len_ratio': float(np.mean(lr))}) print(f" {tag:<34} dtw_h {rows[-1]['dtw_hands']:.4f} FGD {rows[-1]['FGD']:7.3f}" f" MAEJ* {rows[-1]['MAEJ_x100']:7.3f} len {rows[-1]['len_ratio']:.3f}") rows.sort(key=lambda r: r['dtw_hands']) print(f"\n{'condition':<34}{'DTW_h':>9}{'FGD':>9}{'MAEJ*':>9}{'len_r':>8}") print(f"{'real vs real (FGD floor)':<34}{0.0:>9.4f}{floor:>9.3f}{0.0:>9.3f}{1.0:>8.3f}") for r in rows: print(f"{r['condition']:<34}{r['dtw_hands']:>9.4f}{r['FGD']:>9.3f}" f"{r['MAEJ_x100']:>9.3f}{r['len_ratio']:>8.3f}") os.makedirs(os.path.dirname(args.out_json) or '.', exist_ok=True) with open(args.out_json, 'w') as f: json.dump({'protocol': '200 clips / 50 joints / per-clip shoulder-width', 'n_clips': len(names), 'fgd_noise_floor': floor, 'rows': rows}, f, indent=2) print(f'\nwrote {args.out_json}') if __name__ == '__main__': main()