| |
| """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) |
|
|
| |
| 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 |
| 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') |
|
|
| |
| 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} |
|
|
| |
| 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)) |
|
|
| |
| 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() |
|
|
