#!/usr/bin/env python3 """actaug_core -- self-contained helpers for object-perturbed RoboCasa rollouts. Handoff module: imports ONLY installed libraries (robocasa/robosuite/robomimic/ numpy/pandas/zmq/torch/imageio). No imports from any other project directory. Run with the sim interpreter: /lp-dev/jonghoon/mimicgen_augment/envs/mimicgen/bin/python and MUJOCO_GL=egl MUJOCO_EGL_DEVICE_ID=. Covers: * loading one episode of a robocasa LeRobot export (model.xml.gz + states.npz + ep_meta.json under extras/, 12-d actions from data/chunk-*/episode_*.parquet) * building the env from the export's env_args and resetting to the episode * perturbing the manipulated object's initial pose (world-z yaw + xy translation) * GR00T-N1.5 policy client (ZMQ torch.save protocol) + obs/action converters """ import gzip import io import json from pathlib import Path import numpy as np DEFAULT_DATASET = Path( "/lp-dev/jonghoon/robocasa_full/pickplace_target_human/PickPlaceCounterToCabinet") CAMS = ["robot0_agentview_left", "robot0_agentview_right", "robot0_eye_in_hand"] VIEW_KEYS = ["video.left_view", "video.right_view", "video.wrist_view"] IMG = 256 LIFT_M = 0.03 # object raised this much above its start height counts as lifted OBJ_JOINT = "obj_joint0" ACTION_SEED_KEY = "_policy_action_seed" # honored by myGR00T policy.get_action _ENV_CACHE = {} # ---------------------------------------------------------------- episode I/O def load_episode(dataset_root, ep_idx): """Read everything needed to re-simulate episode `ep_idx` of a robocasa LeRobot export. Returns a dict; heavy arrays are numpy.""" root = Path(dataset_root) ed = root / "extras" / f"episode_{ep_idx:06d}" xml = gzip.open(ed / "model.xml.gz", "rt").read() states = np.load(ed / "states.npz")["states"] ep_meta = json.load(open(ed / "ep_meta.json")) import pandas as pd chunk = ep_idx // 1000 pq = root / "data" / f"chunk-{chunk:03d}" / f"episode_{ep_idx:06d}.parquet" df = pd.read_parquet(pq) actions = np.stack(df["action"].to_numpy()).astype(np.float64) # (T, 12) parquet order # instruction: tasks.jsonl row addressed by the annotation column (fallback: ep_meta lang) instr = ep_meta.get("lang", "") try: tasks = {} for line in open(root / "meta" / "tasks.jsonl"): row = json.loads(line) tasks[int(row["task_index"])] = row["task"] instr = tasks[int(df["annotation.human.task_description"].iloc[0])] except Exception: pass env_args = json.load(open(root / "extras" / "dataset_meta.json"))["env_args"] return dict(ep_idx=ep_idx, xml=xml, states=states, ep_meta=ep_meta, actions=actions, instruction=instr, env_args=env_args) def make_env(env_args): """EnvRobosuite for the export's env_args (cached per env_name).""" name = env_args["env_name"] if name in _ENV_CACHE: return _ENV_CACHE[name] import robocasa # noqa: F401 (registers kitchen envs) import robomimic.utils.obs_utils as ObsUtils ObsUtils.initialize_obs_utils_with_obs_specs({"obs": {"low_dim": [], "rgb": []}}) from robomimic.envs.env_robosuite import EnvRobosuite kw = dict(env_args["env_kwargs"]) kw.pop("env_name", None) env = EnvRobosuite(name, render=False, render_offscreen=True, use_image_obs=False, camera_names=CAMS, camera_heights=IMG, camera_widths=IMG, **kw) _ENV_CACHE[name] = env return env def reset_episode(env, ep): """Reset env to the recorded initial sim state of `ep` (from load_episode).""" env.reset_to({"model": ep["xml"], "states": ep["states"][0], "ep_meta": json.dumps(ep["ep_meta"])}) # ------------------------------------------------------------- perturbation def _q_mul(a, b): # wxyz w0, x0, y0, z0 = a w1, x1, y1, z1 = b return np.array([w0*w1 - x0*x1 - y0*y1 - z0*z1, w0*x1 + x0*w1 + y0*z1 - z0*y1, w0*y1 - x0*z1 + y0*w1 + z0*x1, w0*z1 + x0*y1 - y0*x1 + z0*w1]) def apply_object_perturbation(env, dyaw_deg=0.0, dx=0.0, dy=0.0, dz=0.0): """Spin the manipulated object about the WORLD vertical and translate it in the world frame, in-sim, right after reset. Returns (pos_before, pos_after).""" sim = env.env.sim jid = sim.model.joint_name2id(OBJ_JOINT) adr = int(sim.model.jnt_qposadr[jid]) p0 = sim.data.qpos[adr:adr + 3].copy() th = np.radians(dyaw_deg) / 2.0 qz = np.array([np.cos(th), 0.0, 0.0, np.sin(th)]) q = _q_mul(qz, sim.data.qpos[adr + 3:adr + 7].copy()) sim.data.qpos[adr + 3:adr + 7] = q / np.linalg.norm(q) sim.data.qpos[adr:adr + 3] = p0 + np.array([dx, dy, dz]) sim.forward() return p0, sim.data.qpos[adr:adr + 3].copy() def sample_perturbation(rng, rot_deg=180.0, trans_m=0.05, trans_min_m=0.0, rot_min_deg=0.0): """Yaw magnitude uniform in [rot_min_deg, rot_deg] with random sign; translation direction uniform on the circle, radius uniform in [trans_min_m, trans_m]. Returns (dyaw, dx, dy).""" dyaw = float(rng.uniform(rot_min_deg, rot_deg)) * (1 if rng.random() < 0.5 else -1) r = float(rng.uniform(trans_min_m, trans_m)) phi = float(rng.uniform(0, 2 * np.pi)) return dyaw, r * np.cos(phi), r * np.sin(phi) def settle(env, n_steps=10): """Let physics settle after a perturbation (object may be slightly off the counter surface after a yaw about its joint origin).""" sim = env.env.sim for _ in range(n_steps): sim.step() sim.forward() # ------------------------------------------------------ actions & obs bridges def to_env_action(a): """parquet 12-d (base_motion[0:4], control_mode[4], ee_pos[5:8], ee_rot[8:11], grip[11]) -> robosuite env-order 12-d.""" e = np.zeros(12) e[0:3] = a[5:8] e[3:6] = a[8:11] e[6] = 1.0 if a[11] > 0 else -1.0 e[7:11] = a[0:4] e[11] = 1.0 if a[4] > 0 else -1.0 return e def env_to_parquet_action(e): """Inverse of to_env_action (for dumping executed trajectories).""" a = np.zeros(12) a[0:4] = e[7:11] a[4] = 1.0 if e[11] > 0 else -1.0 a[5:8] = e[0:3] a[8:11] = e[3:6] a[11] = 1.0 if e[6] > 0 else -1.0 return a def policy_chunk_to_env(chunk, j): """GR00T action-chunk step j -> env-order 12-d (binarized gripper/control_mode).""" eep = np.asarray(chunk["action.end_effector_position"])[j].ravel()[:3] eer = np.asarray(chunk["action.end_effector_rotation"])[j].ravel()[:3] grip = float(np.asarray(chunk["action.gripper_close"])[j].ravel()[0]) base = np.asarray(chunk["action.base_motion"])[j].ravel()[:4] ctrl = float(np.asarray(chunk["action.control_mode"])[j].ravel()[0]) e = np.zeros(12) e[0:3] = eep e[3:6] = eer e[6] = 1.0 if grip > 0 else -1.0 e[7:11] = base e[11] = 1.0 if ctrl > 0 else -1.0 return e def grasp_frame(actions): """First control step whose recorded gripper bit commands CLOSE.""" close = np.where(actions[:, 11] > 0)[0] return int(close[0]) if len(close) else len(actions) // 2 def render3(env): sim = env.env.sim return [sim.render(camera_name=c, width=IMG, height=IMG)[::-1] for c in CAMS] def groot_state_vec(env): """16-d observation.state in the export's modality.json layout.""" di = env.env._get_observations(force_update=False) return np.concatenate([ np.asarray(di["robot0_base_pos"]).ravel()[:3], np.asarray(di["robot0_base_quat"]).ravel()[:4], np.asarray(di["robot0_base_to_eef_pos"]).ravel()[:3], np.asarray(di["robot0_base_to_eef_quat"]).ravel()[:4], np.asarray(di["robot0_gripper_qpos"]).ravel()[:2], ]).astype(np.float64) def groot_obs(env, instruction, action_seed=None): di = env.env._get_observations(force_update=True) l, r, w = render3(env) obs = {"video.left_view": l[None], "video.right_view": r[None], "video.wrist_view": w[None], "state.end_effector_position_relative": np.asarray(di["robot0_base_to_eef_pos"])[None], "state.end_effector_rotation_relative": np.asarray(di["robot0_base_to_eef_quat"])[None], "state.gripper_qpos": np.asarray(di["robot0_gripper_qpos"])[None], "state.base_position": np.asarray(di["robot0_base_pos"])[None], "state.base_rotation": np.asarray(di["robot0_base_quat"])[None], "annotation.human.action.task_description": [instruction]} if action_seed is not None: obs[ACTION_SEED_KEY] = int(action_seed) return obs def groot_obs_oxe(env, instruction, action_seed=None): """DROID/oxe_droid-format obs for the BASE (non-finetuned) GR00T-N1.5, whose only viable manipulation head here is oxe_droid (best-effort frame mapping, same approach as gripper_state_replay.groot_obs_oxe).""" import robosuite.utils.transform_utils as T di = env.env._get_observations(force_update=True) l, r, w = render3(env) quat = np.asarray(di["robot0_base_to_eef_quat"]) # [x,y,z,w] euler = T.mat2euler(T.quat2mat(quat)) gq = np.asarray(di["robot0_gripper_qpos"]) g01 = float(np.clip(np.mean(np.abs(gq)) / 0.04, 0.0, 1.0)) # 1=open, 0=closed obs = {"video.exterior_image_1": l[None], "video.exterior_image_2": r[None], "video.wrist_image": w[None], "state.eef_position": np.asarray(di["robot0_base_to_eef_pos"])[None], "state.eef_rotation": np.asarray(euler)[None], "state.gripper_position": np.array([[g01]]), "annotation.language.language_instruction": [instruction]} if action_seed is not None: obs[ACTION_SEED_KEY] = int(action_seed) return obs def oxe_chunk_to_env(chunk, j, grip_flip=False): """oxe_droid action step -> robocasa env-order 12-d (base=0, control_mode=-1). DROID gripper_position ~[0,1]; default close = g > 0.5 (flip to invert).""" dp = np.asarray(chunk["action.eef_position_delta"])[j].ravel()[:3] dr = np.asarray(chunk["action.eef_rotation_delta"])[j].ravel()[:3] g = float(np.asarray(chunk["action.gripper_position"])[j].ravel()[0]) e = np.zeros(12) e[0:3] = dp e[3:6] = dr close = (g < 0.5) if grip_flip else (g > 0.5) e[6] = 1.0 if close else -1.0 e[11] = -1.0 return e def obj_z(env): return float(env.env._get_observations(force_update=False)["obj_pos"][2]) def is_success(env): return bool(env.is_success().get("task", False)) # ------------------------------------------------------------- policy client class PolicyClient: """ZMQ REQ client for the myGR00T RobotInferenceServer (torch.save protocol).""" def __init__(self, host="127.0.0.1", port=8801): import zmq self.ctx = zmq.Context() self.sock = self.ctx.socket(zmq.REQ) self.sock.connect(f"tcp://{host}:{port}") def get_action(self, obs): import torch buf = io.BytesIO() torch.save({"endpoint": "get_action", "data": obs}, buf) self.sock.send(buf.getvalue()) msg = self.sock.recv() if msg == b"ERROR": raise RuntimeError("policy server returned ERROR") return torch.load(io.BytesIO(msg), weights_only=False) # ------------------------------------------------------------------- output def write_mp4(path, frames, fps=20): import imageio w = imageio.get_writer(str(path), fps=fps, codec="libx264", macro_block_size=1, ffmpeg_params=["-pix_fmt", "yuv420p"]) for f in frames: w.append_data(np.asarray(f).astype(np.uint8)) w.close()