o vi: @sdZddlZddlZddlZdedededejfddZd ejd ed ededejf d d Z de de fddZ dededededef ddZ de de fddZde de fddZde fddZd ed!ed"ede fd#d$ZdS)%z4 Common helper functions for robot dataset classes. N num_stepsterminal_rewardgammareturncCs|}tj|tjd}||d<t|}d}tt|D]}||||}|||<q|dkr8d||d}|Sd|d}|S)a Compute Monte Carlo returns with sparse reward at the end of an episode. Returns are rescaled from [0, terminal_reward] to [-1, 1]. Args: num_steps (int): Number of steps in the episode terminal_reward (float): Reward at the terminal state (usually 0 or 1) gamma (float): Discount factor Returns: np.ndarray: Returns array of shape (num_steps,) rescaled to [-1, 1] )dtypegr)npzerosfloat32 zeros_likereversedrange)rrrTZrewardsreturnsGtrK/data/cameron/vidgen/cosmos-policy/cosmos_policy/datasets/dataset_common.pycompute_monte_carlo_returnss    ractionsrelative_step_idx chunk_sizec Csb||}||kr||||}|S||d}||}t|d|df}tj||gdd}|S)a* Get action chunk starting from relative_step_idx with padding if needed. If there aren't enough remaining actions to fill the chunk, the last action is repeated to fill the chunk. Args: actions (np.ndarray): Full actions array of shape (num_steps, action_dim) relative_step_idx (int): Starting index for the action chunk chunk_size (int): Size of the action chunk num_steps (int): Total number of steps in the episode Returns: np.ndarray: Action chunk of shape (chunk_size, action_dim) Nrr r)axis)r tile concatenate) rrrrZremaining_actions action_chunkZavailable_actionsZnum_padding_neededpaddingrrrget_action_chunk_with_padding:s r rollout_datarollout_episode_metadatac Csd}i}i}d}d}|D].\}}|d} |d} t| D]} | r-|| f||<|d7}n || f||<|d7}|d7}qq|D]2\}} t| d} t| d} t| D]} | rd|| f||<|d7}n || f||<|d7}|d7}qUqA|||||dS)a Build mapping for rollout dataset with separate tracking for successful/failure episodes. This function creates three mappings: - _rollout_success_step_to_episode_map: Maps global success step index to (episode_idx, relative_idx) - _rollout_failure_step_to_episode_map: Maps global failure step index to (episode_idx, relative_idx) - Counters for total steps in each category Args: rollout_data (dict): In-memory rollout data (episode_idx -> episode_data) rollout_episode_metadata (dict): Lazy-loaded rollout metadata (episode_idx -> metadata) Returns: dict: Dictionary containing: - '_rollout_success_step_to_episode_map': dict mapping success step indices - '_rollout_failure_step_to_episode_map': dict mapping failure step indices - '_rollout_success_total_steps': int - '_rollout_failure_total_steps': int - '_rollout_total_steps': int rrsuccessr )$_rollout_success_step_to_episode_map$_rollout_failure_step_to_episode_map_rollout_success_total_steps_rollout_failure_total_steps_rollout_total_steps)itemsrintbool) r r!rollout_total_stepsZ#rollout_success_step_to_episode_mapZ#rollout_failure_step_to_episode_maprollout_success_total_stepsrollout_failure_total_steps episode_idx episode_datar is_successiepisode_metadatarrr build_rollout_step_index_mappingbsB             r3r,r-demonstration_sampling_probsuccess_rollout_sampling_probcCsP||}|dk}|r|dks|dkrd}|}d}n4|dks!|dkr(|}d}d}n%|} |} t| |d|} t| d||} | | krI| }| }n| }| }|dksU|dkrXd} nC|dkrc|} d}d}n8|}||}t||d|}t|d||}||kr|} n|} t||}t|d|}n|} d}d}| ||}| |||dS)ai Calculate epoch layout with proper scaling: demos, success rollouts, failure rollouts. This function computes adjusted counts for each data type to maintain the desired sampling probabilities during training. Args: num_steps (int): Number of demonstration steps rollout_success_total_steps (int): Number of success rollout steps rollout_failure_total_steps (int): Number of failure rollout steps demonstration_sampling_prob (float): Target probability of sampling demos vs rollouts success_rollout_sampling_prob (float): Target probability of sampling success vs failure rollouts Returns: dict: Dictionary containing: - 'adjusted_demo_count': int - 'adjusted_success_rollout_count': int - 'adjusted_failure_rollout_count': int - 'epoch_length': int rr )adjusted_demo_countadjusted_success_rollout_countadjusted_failure_rollout_count epoch_length)r))rr,r-r4r5r+Zhas_rollout_datar7r8sfZs_newf_newr6drZd_newZr_newr9rrrcalculate_epoch_structuresX  r?data_dirdatac Ctj|d}tj|r/t|d }t|}Wdn1s"wYtd|n9||}i}|D] \}}| ||<q9t|d}tj ||ddWdn1s\wYtd|i}|D] \}}t |||<qn|S) a Load dataset statistics from JSON file if it exists, otherwise compute and save them. Args: data_dir (str): Directory where statistics JSON file should be stored data (dict): Dataset dictionary (episode_idx -> episode_data) calculate_dataset_statistics_func: Function to compute statistics from data Returns: dict: Dataset statistics with numpy array values zdataset_statistics.jsonr>Nz Loaded dataset statistics from: windentzDataset statistics saved to: ospathjoinexistsopenjsonloadprintr(tolistdumpr array) r@rA!calculate_dataset_statistics_funcdataset_stats_pathr; json_stats dataset_stats stat_name stat_valuerrr"load_or_compute_dataset_statistics$    rYc CrB) a Load post-normalization dataset statistics from JSON file if it exists, otherwise compute and save them. Args: data_dir (str): Directory where statistics JSON file should be stored data (dict): Dataset dictionary (episode_idx -> episode_data) with normalized data calculate_dataset_statistics_func: Function to compute statistics from data Returns: dict: Post-normalization dataset statistics with numpy array values z!dataset_statistics_post_norm.jsonr>Nz3Loaded post-normalization dataset statistics from: rCrDrEz0Post-normalization dataset statistics saved to: rG) r@rArSZdataset_stats_post_norm_pathr;rUdataset_stats_post_normrWrXrrr-load_or_compute_post_normalization_statistics rZr\cCsNi}d}|D]\}}|d}t|D] }||f||<|d7}qq||dS)am Build a mapping from global step index to (episode index, relative index within episode). Args: data (dict): Dataset dictionary (episode_idx -> episode_data) Returns: dict: Dictionary containing: - '_step_to_episode_map': dict mapping global step index to (episode_idx, relative_idx) - '_total_steps': int rrr )_step_to_episode_map _total_steps)r(r)rAZstep_to_episode_map total_stepsr.r/rr1rrrbuild_demo_step_index_mappingFs    r`idxr6r7cCs ||krdS|||krdSdS)a Determine which dataset to sample from based on index ranges. Layout of indices: [demos] [success rollouts] [failure rollouts] Args: idx (int): Sample index adjusted_demo_count (int): Number of demo samples in epoch adjusted_success_rollout_count (int): Number of success rollout samples in epoch Returns: str: One of "demo", "success_rollout", or "failure_rollout" demosuccess_rolloutfailure_rolloutr)rar6r7rrrdetermine_sample_type`s  re)__doc__rMrHnumpyr r)floatndarrayrrdictr3r?strrYr\r`rerrrrs@! (C U&&