Humanoid Bimanual Teleoperation Data

import h5py
import numpy as np

# Load Blue Projects Humanoid Bimanual Teleoperation Episode File
def load_bimanual_episode(h5_file_path: str):
    """
    Parses 50Hz bimanual telemetry dataset container.
    Returns joint angles (qpos), velocities (qvel), and 6-axis wrench vectors.
    """
    with h5py.File(h5_file_path, 'r') as f:
        # Extract observations matrix (50Hz timestep rate)
        joint_angles = np.array(f['observations/qpos'], dtype=np.float32)       # Shape: [T, 14]
        joint_velocities = np.array(f['observations/qvel'], dtype=np.float32)   # Shape: [T, 14]
        force_torque = np.array(f['observations/force_torque'], dtype=np.float32) # Shape: [T, 6] (Fx,Fy,Fz,Tx,Ty,Tz)
        actions = np.array(f['action'], dtype=np.float32)                        # Target joint commands [T, 14]

        print(f"Loaded Episode: {h5_file_path}")
        print(f"  - Timesteps (T): {len(joint_angles)} frames ({len(joint_angles)*0.020:.2f}s)")
        print(f"  - Joint Angles Shape: {joint_angles.shape}")
        print(f"  - Force-Torque Wrench Range: {force_torque.min():.2f}N to {force_torque.max():.2f}N")

        return {
            'qpos': joint_angles,
            'qvel': joint_velocities,
            'force_torque': force_torque,
            'action': actions
        }

if __name__ == '__main__':
    data = load_bimanual_episode('humanoid_teleop_episode_042.h5')

import h5py
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader

class HumanoidBimanualDataset(Dataset):
    """
    Production PyTorch Dataset loader for Blue Projects 50Hz Bimanual Teleoperation HDF5 datasets.
    Designed for ACT (Action Chunking with Transformers), Diffusion Policy, and OpenVLA architectures.
    """
    def __init__(self, hdf5_path: str, chunk_size: int = 16):
        self.hdf5_path = hdf5_path
        self.chunk_size = chunk_size
        
        with h5py.File(self.hdf5_path, 'r') as root:
            assert 'observations/qpos' in root, "Missing qpos joint observations"
            assert 'observations/qvel' in root, "Missing qvel velocity observations"
            assert 'observations/force_torque' in root, "Missing 6-axis load cell data"
            
            self.qpos = np.array(root['observations/qpos'], dtype=np.float32)          # [T, 14]
            self.qvel = np.array(root['observations/qvel'], dtype=np.float32)          # [T, 14]
            self.force_torque = np.array(root['observations/force_torque'], dtype=np.float32) # [T, 6]
            self.actions = np.array(root['action'], dtype=np.float32)                   # [T, 14]
            
            self.total_timesteps = self.qpos.shape[0]

    def __len__(self):
        return max(1, self.total_timesteps - self.chunk_size + 1)

    def __getitem__(self, idx: int):
        end_idx = idx + self.chunk_size
        obs_state = torch.tensor(np.concatenate([self.qpos[idx], self.qvel[idx], self.force_torque[idx]], axis=-1))
        action_chunk = torch.tensor(self.actions[idx:end_idx])
        
        return {
            'observation/state': obs_state,   # 14 + 14 + 6 = 34 dimensions
            'action_chunk': action_chunk       # [chunk_size, 14]
        }

if __name__ == '__main__':
    dataset = HumanoidBimanualDataset('humanoid_teleop_episode_042.h5', chunk_size=16)
    loader = DataLoader(dataset, batch_size=32, shuffle=True)
    for batch in loader:
        print("Batch State Shape:", batch['observation/state'].shape) # [32, 34]
        print("Action Chunk Shape:", batch['action_chunk'].shape)    # [32, 16, 14]
        break