| |
| """Push-Box BOX+MARK point-flow overlay: GT tracks vs label-conditioned 3-PoinTr pred. |
| |
| Uses the two-object label-conditioned model (push_box_boxmark_150pts_384), tracking |
| BOTH the cheez-it box (label 0) and the red-cross mark (label 1). |
| |
| GREEN = GT CoTracker tracks (box+mark). RED = 3-PoinTr predicted flow from frame-0 |
| GT points + their labels. Tracks live in 384 space (the videos_small domain they |
| were built on); we draw on each demo's own input.mp4 (384), upscaled 2x for clarity. |
| |
| Usage: python gen_pred_flow_pushbox_boxmark.py --ep ep000 [--ep ep005 ...] |
| """ |
| import argparse, os, sys |
| import cv2, imageio |
| import numpy as np |
| import torch |
|
|
| POINTR_REPO = "/home/hongyi/PWM/reward_model/3PoinTr_repo" |
| if POINTR_REPO not in sys.path: |
| sys.path.insert(0, POINTR_REPO) |
| from util.training_utils.arch import TrajectoryModel |
| from util.geometry_utils.geom_utils import ( |
| compute_start_sphere_params, sim_transform, sim_transform_inverse_points, |
| ) |
|
|
| CKPT = "/home/hongyi/PWM/reward_model/3pointr_models/push_box_boxmark_150pts_384_flow_backbone_final.pth" |
| TRACKS_ROOT = "/data0/hongyi/reward_model_data/3pointr_data_push_box_2d_boxmark" |
| SQUASH = 384 |
| UPSCALE = 2 |
| MAX_POINTS = 150 |
|
|
|
|
| def infer_horizon(state): |
| w = state.get("decoder_only.point_flow_head.weight") |
| return None if w is None else int(w.shape[0] // 2) |
|
|
|
|
| def load_model(ckpt_path, device): |
| ckpt = torch.load(ckpt_path, map_location="cpu", weights_only=False) |
| cfg = ckpt.get("args", {}) or {} |
| state = ckpt.get("flow_backbone_state_dict") or ckpt.get("model_state_dict") or ckpt |
| horizon = infer_horizon(state) or int(cfg.get("horizon", 306)) |
| use_labels = any("point_label_embed" in k for k in state.keys()) |
| model = TrajectoryModel( |
| arch=cfg.get("arch", "flow_conditioned"), |
| dim=int(cfg.get("arch_dim", 256)), depth=int(cfg.get("arch_depth", 2)), |
| query_depth=int(cfg.get("arch_query_depth", 1)), |
| num_heads=int(cfg.get("arch_heads", 4)), |
| query_heads=int(cfg.get("arch_query_heads", 1)), |
| use_task_conditioning=False, num_tasks=1, horizon=horizon, |
| num_queries=int(cfg.get("num_queries", 10)), point_dim=2, |
| use_point_labels=use_labels, |
| ).to(device) |
| missing, unexpected = model.load_state_dict(state, strict=False) |
| print(f"[predict] horizon={horizon} use_point_labels={use_labels} " |
| f"missing={len(missing)} unexpected={len(unexpected)}") |
| model.eval() |
| return model, horizon |
|
|
|
|
| @torch.no_grad() |
| def predict(model, horizon, x0_px, labels, device): |
| x0 = torch.from_numpy(x0_px.astype(np.float32)).to(device).unsqueeze(0) |
| scales, center = compute_start_sphere_params(x0) |
| inv_scales = 1.0 / scales |
| x0n = sim_transform(x0, scales=inv_scales, rotations=None, translations=-center) |
| batch = {"x0": x0n, "num_steps": horizon, |
| "task_ids": torch.zeros(1, dtype=torch.long, device=device), |
| "scales": scales, "center": center, |
| "point_labels": torch.from_numpy(labels.astype(np.int64)).to(device).unsqueeze(0)} |
| out = model(batch, mode="eval") |
| pred = out["pred_pos"] |
| L, N = pred.shape[1], pred.shape[2] |
| un = sim_transform_inverse_points(pred.reshape(1, L * N, 2), |
| scales=inv_scales, rotations=None, translations=-center) |
| return un.view(L, N, 2).cpu().numpy() |
|
|
|
|
| def render(ep, model, horizon, device, out_dir): |
| d = f"{TRACKS_ROOT}/tableHF_{ep}" |
| z = np.load(f"{d}/tracks_labeled.npz", allow_pickle=True) |
| tr = z["tracks"].astype(np.float32) |
| lab = z["labels"].astype(np.int64) |
| vis = z["vis"] if "vis" in z.files else np.ones(tr.shape[:2], bool) |
| N = tr.shape[1] |
| if N > MAX_POINTS: |
| idx = np.linspace(0, N - 1, MAX_POINTS).astype(int) |
| tr, vis, lab = tr[:, idx], vis[:, idx], lab[idx] |
| N = tr.shape[1] |
|
|
| cap = cv2.VideoCapture(f"{d}/input.mp4") |
| frames = [] |
| while True: |
| ok, f = cap.read() |
| if not ok: |
| break |
| f = cv2.cvtColor(f, cv2.COLOR_BGR2RGB) |
| if UPSCALE != 1: |
| f = cv2.resize(f, (SQUASH * UPSCALE, SQUASH * UPSCALE), interpolation=cv2.INTER_LINEAR) |
| frames.append(f) |
| cap.release() |
| Td = min(len(frames), tr.shape[0]) |
| H, W = frames[0].shape[:2] |
| s = float(UPSCALE) |
|
|
| pred384 = predict(model, horizon, tr[0], lab, device) |
| pred = pred384 * s |
| gt = tr * s |
| L = pred.shape[0] |
|
|
| tt = np.linspace(0, L - 1, Td, dtype=np.float32) |
| t0 = np.floor(tt).astype(int).clip(0, L - 1); t1 = np.minimum(t0 + 1, L - 1) |
| w = (tt - t0)[:, None, None] |
| pred_al = (1 - w) * pred[t0] + w * pred[t1] |
|
|
| out_frames = [] |
| for ti in range(Td): |
| cur = frames[ti].copy() |
| for pi in range(N): |
| if vis[ti, pi]: |
| x, y = gt[ti, pi]; cv2.circle(cur, (int(x), int(y)), 2, (0, 255, 0), -1) |
| for pi in range(N): |
| x, y = pred_al[ti, pi]; cv2.circle(cur, (int(x), int(y)), 2, (255, 0, 0), -1) |
| err = np.linalg.norm(gt[ti] - pred_al[ti], axis=-1) |
| m = vis[ti].astype(np.float32); me = float((err * m).sum() / max(m.sum(), 1)) / s |
| cv2.putText(cur, f"{ep} t={ti} mean err={me:4.1f}px@384", (10, 26), |
| cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 0), 2) |
| cv2.putText(cur, "GT box+mark", (10, H - 34), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2) |
| cv2.putText(cur, "3PoinTr pred", (10, H - 12), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 0, 0), 2) |
| out_frames.append(cur) |
|
|
| os.makedirs(out_dir, exist_ok=True) |
| out = f"{out_dir}/pred_flow_{ep}.mp4" |
| imageio.mimwrite(out, out_frames, fps=15, macro_block_size=1) |
| err = np.linalg.norm(gt[:Td] - pred_al, axis=-1) * vis[:Td] |
| me = (err.sum(1) / vis[:Td].sum(1).clip(min=1)) / s |
| nb = int((lab == 0).sum()); nm = int((lab == 1).sum()) |
| print(f"[{ep}] N={N} (box={nb} mark={nm}) L={L} Td={Td} px err@384 " |
| f"first={me[0]:.1f} mid={me[Td//2]:.1f} last={me[-1]:.1f} mean={me.mean():.1f} -> {out}") |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--ep", action="append", default=[]) |
| ap.add_argument("--ckpt", default=CKPT) |
| ap.add_argument("--out_dir", default="/home/hongyi/PWM/external/pred_flow_pushbox_boxmark") |
| ap.add_argument("--device", default="cuda:0") |
| a = ap.parse_args() |
| eps = a.ep or ["ep000"] |
| device = torch.device(a.device) |
| model, horizon = load_model(a.ckpt, device) |
| for ep in eps: |
| try: |
| render(ep, model, horizon, device, a.out_dir) |
| except Exception as e: |
| print(f"[{ep}] FAILED: {e}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
