from functools import partial import json import os import re import time from enum import IntEnum from typing import Tuple import chex import jax import jax.numpy as jnp import numpy as np import optax import orbax.checkpoint as ocp from flax import core, struct from flax.training.train_state import TrainState as BaseTrainState import wandb from jaxued.environments.underspecified_env import EnvParams, EnvState, Observation, UnderspecifiedEnv from jaxued.level_sampler import LevelSampler from jaxued.utils import compute_max_returns, max_mc, positive_value_loss from kinetix.environment.env import PixelObservations, make_kinetix_env_from_name from kinetix.environment.env_state import StaticEnvParams from kinetix.environment.utils import permute_pcg_state from kinetix.environment.wrappers import ( UnderspecifiedToGymnaxWrapper, LogWrapper, DenseRewardWrapper, AutoReplayWrapper, ) from kinetix.models import make_network_from_config from kinetix.pcg.pcg import env_state_to_pcg_state from kinetix.render.renderer_pixels import make_render_pixels from kinetix.models.actor_critic import ScannedRNN from kinetix.util.saving import ( expand_pcg_state, get_pcg_state_from_json, load_pcg_state_pickle, load_world_state_pickle, stack_list_of_pytrees, import_env_state_from_json, load_from_json_file, ) from flax.training.train_state import TrainState BASE_DIR = "worlds" DEFAULT_EVAL_LEVELS = [ "easy.cartpole", "easy.flappy_bird", "easy.unicycle", "easy.car_left", "easy.car_right", "easy.pinball", "easy.swing_up", "easy.thruster", ] def get_eval_levels(eval_levels, static_env_params): should_permute = [".permute" in l for l in eval_levels] eval_levels = [re.sub(r"\.permute\d+", "", l) for l in eval_levels] ls = [get_pcg_state_from_json(os.path.join(BASE_DIR, l + ("" if l.endswith(".json") else ".json"))) for l in eval_levels] ls = [expand_pcg_state(l, static_env_params) for l in ls] new_ls = [] rng = jax.random.PRNGKey(0) for sp, l in zip(should_permute, ls): rng, _rng = jax.random.split(rng) if sp: l = permute_pcg_state(_rng, l, static_env_params) new_ls.append(l) return stack_list_of_pytrees(new_ls) def evaluate_rnn( # from jaxued rng: chex.PRNGKey, env: UnderspecifiedEnv, env_params: EnvParams, train_state: TrainState, init_hstate: chex.ArrayTree, init_obs: Observation, init_env_state: EnvState, max_episode_length: int, keep_states=True, return_trajectories=False, ) -> Tuple[chex.Array, chex.Array, chex.Array]: """This runs the RNN on the environment, given an initial state and observation, and returns (states, rewards, episode_lengths) Args: rng (chex.PRNGKey): env (UnderspecifiedEnv): env_params (EnvParams): train_state (TrainState): init_hstate (chex.ArrayTree): Shape (num_levels, ) init_obs (Observation): Shape (num_levels, ) init_env_state (EnvState): Shape (num_levels, ) max_episode_length (int): Returns: Tuple[chex.Array, chex.Array, chex.Array]: (States, rewards, episode lengths) ((NUM_STEPS, NUM_LEVELS), (NUM_STEPS, NUM_LEVELS), (NUM_LEVELS,) """ num_levels = jax.tree_util.tree_flatten(init_obs)[0][0].shape[0] def step(carry, _): rng, hstate, obs, state, done, mask, episode_length = carry rng, rng_action, rng_step = jax.random.split(rng, 3) x = jax.tree.map(lambda x: x[None, ...], (obs, done)) hstate, pi, _ = train_state.apply_fn(train_state.params, hstate, x) action = pi.sample(seed=rng_action).squeeze(0) obs, next_state, reward, done, info = jax.vmap(env.step, in_axes=(0, 0, 0, None))( jax.random.split(rng_step, num_levels), state, action, env_params ) next_mask = mask & ~done episode_length += mask if keep_states: return (rng, hstate, obs, next_state, done, next_mask, episode_length), (state, reward, done, info) else: return (rng, hstate, obs, next_state, done, next_mask, episode_length), (None, reward, done, info) (_, _, _, _, _, _, episode_lengths), (states, rewards, dones, infos) = jax.lax.scan( step, ( rng, init_hstate, init_obs, init_env_state, jnp.zeros(num_levels, dtype=bool), jnp.ones(num_levels, dtype=bool), jnp.zeros(num_levels, dtype=jnp.int32), ), None, length=max_episode_length, ) done_idx = jnp.argmax(dones, axis=0) to_return = (states, rewards, done_idx, episode_lengths, infos) if return_trajectories: return to_return, (dones, rewards) return to_return def general_eval( rng: chex.PRNGKey, eval_env: UnderspecifiedEnv, env_params: EnvParams, train_state: TrainState, levels: EnvState, num_eval_steps: int, num_levels: int, keep_states=True, return_trajectories=False, ): """ This evaluates the current policy on the set of evaluation levels It returns (states, cum_rewards, episode_lengths), with shapes (num_steps, num_eval_levels, ...), (num_eval_levels,), (num_eval_levels,) """ rng, rng_reset = jax.random.split(rng) init_obs, init_env_state = jax.vmap(eval_env.reset_to_level, (0, 0, None))( jax.random.split(rng_reset, num_levels), levels, env_params ) init_hstate = ScannedRNN.initialize_carry(num_levels) (states, rewards, done_idx, episode_lengths, infos), (dones, reward) = evaluate_rnn( rng, eval_env, env_params, train_state, init_hstate, init_obs, init_env_state, num_eval_steps, keep_states=keep_states, return_trajectories=True, ) mask = jnp.arange(num_eval_steps)[..., None] < episode_lengths cum_rewards = (rewards * mask).sum(axis=0) to_return = ( states, cum_rewards, done_idx, episode_lengths, infos, ) # (num_steps, num_eval_levels, ...), (num_eval_levels,), (num_eval_levels,) if return_trajectories: return to_return, (dones, reward) return to_return def compute_gae( gamma: float, lambd: float, last_value: chex.Array, values: chex.Array, rewards: chex.Array, dones: chex.Array, ) -> Tuple[chex.Array, chex.Array]: """This takes in arrays of shape (NUM_STEPS, NUM_ENVS) and returns the advantages and targets. Args: gamma (float): lambd (float): last_value (chex.Array): Shape (NUM_ENVS) values (chex.Array): Shape (NUM_STEPS, NUM_ENVS) rewards (chex.Array): Shape (NUM_STEPS, NUM_ENVS) dones (chex.Array): Shape (NUM_STEPS, NUM_ENVS) Returns: Tuple[chex.Array, chex.Array]: advantages, targets; each of shape (NUM_STEPS, NUM_ENVS) """ def compute_gae_at_timestep(carry, x): gae, next_value = carry value, reward, done = x delta = reward + gamma * next_value * (1 - done) - value gae = delta + gamma * lambd * (1 - done) * gae return (gae, value), gae _, advantages = jax.lax.scan( compute_gae_at_timestep, (jnp.zeros_like(last_value), last_value), (values, rewards, dones), reverse=True, unroll=16, ) return advantages, advantages + values def sample_trajectories_rnn( rng: chex.PRNGKey, env: UnderspecifiedEnv, env_params: EnvParams, train_state: TrainState, init_hstate: chex.ArrayTree, init_obs: Observation, init_env_state: EnvState, num_envs: int, max_episode_length: int, return_states: bool = False, ) -> Tuple[ Tuple[chex.PRNGKey, TrainState, chex.ArrayTree, Observation, EnvState, chex.Array], Tuple[Observation, chex.Array, chex.Array, chex.Array, chex.Array, chex.Array, dict], ]: """This samples trajectories from the environment using the agent specified by the `train_state`. Args: rng (chex.PRNGKey): Singleton env (UnderspecifiedEnv): env_params (EnvParams): train_state (TrainState): Singleton init_hstate (chex.ArrayTree): This is the init RNN hidden state, has to have shape (NUM_ENVS, ...) init_obs (Observation): The initial observation, shape (NUM_ENVS, ...) init_env_state (EnvState): The initial env state (NUM_ENVS, ...) num_envs (int): The number of envs that are vmapped over. max_episode_length (int): The maximum episode length, i.e., the number of steps to do the rollouts for. Returns: Tuple[Tuple[chex.PRNGKey, TrainState, chex.ArrayTree, Observation, EnvState, chex.Array], Tuple[Observation, chex.Array, chex.Array, chex.Array, chex.Array, chex.Array, dict]]: (rng, train_state, hstate, last_obs, last_env_state, last_value), traj, where traj is (obs, action, reward, done, log_prob, value, info). The first element in the tuple consists of arrays that have shapes (NUM_ENVS, ...) (except `rng` and and `train_state` which are singleton). The second element in the tuple is of shape (NUM_STEPS, NUM_ENVS, ...), and it contains the trajectory. """ def sample_step(carry, _): rng, train_state, hstate, obs, env_state, last_done = carry prev_state = env_state rng, rng_action, rng_step = jax.random.split(rng, 3) x = jax.tree.map(lambda x: x[None, ...], (obs, last_done)) hstate, pi, value = train_state.apply_fn(train_state.params, hstate, x) action = pi.sample(seed=rng_action) log_prob = pi.log_prob(action) value, action, log_prob = jax.tree.map(lambda x: x.squeeze(0), (value, action, log_prob)) next_obs, env_state, reward, done, info = jax.vmap(env.step, in_axes=(0, 0, 0, None))( jax.random.split(rng_step, num_envs), env_state, action, env_params ) carry = (rng, train_state, hstate, next_obs, env_state, done) step = (obs, action, reward, done, log_prob, value, info) if return_states: step += (prev_state,) return carry, step (rng, train_state, hstate, last_obs, last_env_state, last_done), traj = jax.lax.scan( sample_step, ( rng, train_state, init_hstate, init_obs, init_env_state, jnp.zeros(num_envs, dtype=bool), ), None, length=max_episode_length, ) x = jax.tree.map(lambda x: x[None, ...], (last_obs, last_done)) _, _, last_value = train_state.apply_fn(train_state.params, hstate, x) my_obs = traj[0] rew = traj[2] return (rng, train_state, hstate, last_obs, last_env_state, last_value.squeeze(0)), traj def update_actor_critic_rnn( rng: chex.PRNGKey, train_state: TrainState, init_hstate: chex.ArrayTree, batch: chex.ArrayTree, num_envs: int, n_steps: int, n_minibatch: int, n_epochs: int, clip_eps: float, entropy_coeff: float, critic_coeff: float, update_grad: bool = True, ) -> Tuple[Tuple[chex.PRNGKey, TrainState], chex.ArrayTree]: """This function takes in a rollout, and PPO hyperparameters, and updates the train state. Args: rng (chex.PRNGKey): train_state (TrainState): init_hstate (chex.ArrayTree): batch (chex.ArrayTree): obs, actions, dones, log_probs, values, targets, advantages num_envs (int): n_steps (int): n_minibatch (int): n_epochs (int): clip_eps (float): entropy_coeff (float): critic_coeff (float): update_grad (bool, optional): If False, the train state does not actually get updated. Defaults to True. Returns: Tuple[Tuple[chex.PRNGKey, TrainState], chex.ArrayTree]: It returns a new rng, the updated train_state, and the losses. The losses have structure (loss, (l_vf, l_clip, entropy)) """ obs, actions, dones, log_probs, values, targets, advantages = batch last_dones = jnp.roll(dones, 1, axis=0).at[0].set(False) batch = obs, actions, last_dones, log_probs, values, targets, advantages def update_epoch(carry, _): def update_minibatch(train_state, minibatch): init_hstate, obs, actions, last_dones, log_probs, values, targets, advantages = minibatch def loss_fn(params): _, pi, values_pred = train_state.apply_fn(params, init_hstate, (obs, last_dones)) log_probs_pred = pi.log_prob(actions) entropy = pi.entropy().mean() ratio = jnp.exp(log_probs_pred - log_probs) A = (advantages - advantages.mean()) / (advantages.std() + 1e-8) l_clip = (-jnp.minimum(ratio * A, jnp.clip(ratio, 1 - clip_eps, 1 + clip_eps) * A)).mean() values_pred_clipped = values + (values_pred - values).clip(-clip_eps, clip_eps) l_vf = 0.5 * jnp.maximum((values_pred - targets) ** 2, (values_pred_clipped - targets) ** 2).mean() loss = l_clip + critic_coeff * l_vf - entropy_coeff * entropy return loss, (l_vf, l_clip, entropy) grad_fn = jax.value_and_grad(loss_fn, has_aux=True) loss, grads = grad_fn(train_state.params) if update_grad: train_state = train_state.apply_gradients(grads=grads) grad_norm = jnp.linalg.norm( jnp.concatenate(jax.tree_util.tree_map(lambda x: x.flatten(), jax.tree_util.tree_flatten(grads)[0])) ) return train_state, (loss, grad_norm) rng, train_state = carry rng, rng_perm = jax.random.split(rng) permutation = jax.random.permutation(rng_perm, num_envs) minibatches = ( jax.tree.map( lambda x: jnp.take(x, permutation, axis=0).reshape(n_minibatch, -1, *x.shape[1:]), init_hstate, ), *jax.tree.map( lambda x: jnp.take(x, permutation, axis=1) .reshape(x.shape[0], n_minibatch, -1, *x.shape[2:]) .swapaxes(0, 1), batch, ), ) train_state, (losses, grads) = jax.lax.scan(update_minibatch, train_state, minibatches) return (rng, train_state), (losses, grads) return jax.lax.scan(update_epoch, (rng, train_state), None, n_epochs) @partial(jax.jit, static_argnums=(0, 2, 8, 9)) def sample_trajectories_and_learn( env: UnderspecifiedEnv, env_params: EnvParams, config: dict, rng: chex.PRNGKey, train_state: TrainState, init_hstate: chex.Array, init_obs: Observation, init_env_state: EnvState, update_grad: bool = True, return_states: bool = False, ) -> Tuple[ Tuple[chex.PRNGKey, TrainState, Observation, EnvState], Tuple[ Observation, chex.Array, chex.Array, chex.Array, chex.Array, chex.Array, dict, chex.Array, chex.Array, chex.ArrayTree, chex.Array, ], ]: """This function loops the following: - rollout for config['num_steps'] - learn / update policy And it loops it for config['outer_rollout_steps']. What is returns is a new carry (rng, train_state, init_obs, init_env_state), and concatenated rollouts. The shape of the rollouts are config['num_steps'] * config['outer_rollout_steps']. In other words, the trajectories returned by this function are the same as if we ran rollouts for config['num_steps'] * config['outer_rollout_steps'] steps, but the agent does perform PPO updates in between. Args: env (UnderspecifiedEnv): env_params (EnvParams): config (dict): rng (chex.PRNGKey): train_state (TrainState): init_obs (Observation): init_env_state (EnvState): update_grad (bool, optional): Defaults to True. Returns: Tuple[Tuple[chex.PRNGKey, TrainState, Observation, EnvState], Tuple[Observation, chex.Array, chex.Array, chex.Array, chex.Array, chex.Array, dict, chex.Array, chex.Array, chex.ArrayTree, chex.Array]]: This returns a tuple: ( (rng, train_state, init_obs, init_env_state), (obs, actions, rewards, dones, log_probs, values, info, advantages, targets, losses, grads) ) """ def single_step(carry, _): rng, train_state, init_hstate, init_obs, init_env_state = carry ((rng, train_state, new_hstate, last_obs, last_env_state, last_value), traj,) = sample_trajectories_rnn( rng, env, env_params, train_state, init_hstate, init_obs, init_env_state, config["num_train_envs"], config["num_steps"], return_states=return_states, ) if return_states: states = traj[-1] traj = traj[:-1] (obs, actions, rewards, dones, log_probs, values, info) = traj advantages, targets = compute_gae(config["gamma"], config["gae_lambda"], last_value, values, rewards, dones) # Update the policy using trajectories collected from replay levels (rng, train_state), (losses, grads) = update_actor_critic_rnn( rng, train_state, init_hstate, (obs, actions, dones, log_probs, values, targets, advantages), config["num_train_envs"], config["num_steps"], config["num_minibatches"], config["update_epochs"], config["clip_eps"], config["ent_coef"], config["vf_coef"], update_grad=update_grad, ) new_carry = (rng, train_state, new_hstate, last_obs, last_env_state) step = (obs, actions, rewards, dones, log_probs, values, info, advantages, targets, losses, grads) if return_states: step += (states,) return new_carry, step carry = (rng, train_state, init_hstate, init_obs, init_env_state) new_carry, all_rollouts = jax.lax.scan(single_step, carry, None, length=config["outer_rollout_steps"]) all_rollouts = jax.tree_util.tree_map(lambda x: jnp.concatenate(x, axis=0), all_rollouts) return new_carry, all_rollouts def no_op_rollout( env: UnderspecifiedEnv, env_params: EnvParams, rng: chex.PRNGKey, init_obs: Observation, init_env_state: EnvState, num_envs: int, max_episode_length: int, do_random=False, ): noop = jnp.array(env.action_type.noop_action()) zero_action = jnp.repeat(noop[None, ...], num_envs, axis=0) SHAPE = zero_action.shape def sample_step(carry, _): rng, obs, env_state, last_done = carry rng, rng_step, _rng = jax.random.split(rng, 3) if do_random: action = jax.vmap(env.action_space(env_params).sample)(jax.random.split(_rng, num_envs)) else: action = zero_action next_obs, env_state, reward, done, info = jax.vmap(env.step, in_axes=(0, 0, 0, None))( jax.random.split(rng_step, num_envs), env_state, action, env_params ) carry = (rng, next_obs, env_state, done) return carry, (obs, action, reward, done, info) (rng, last_obs, last_env_state, last_done), traj = jax.lax.scan( sample_step, ( rng, init_obs, init_env_state, jnp.zeros(num_envs, dtype=bool), ), None, length=max_episode_length, ) info = traj[-1] dones = traj[-2] returns_per_env = (info["returned_episode_returns"] * dones).sum(axis=0) / jnp.maximum(1, dones.sum(axis=0)) lens_per_env = (info["returned_episode_lengths"] * dones).sum(axis=0) / jnp.maximum(1, dones.sum(axis=0)) success_per_env = (info["returned_episode_solved"] * dones).sum(axis=0) / jnp.maximum(1, dones.sum(axis=0)) return returns_per_env, lens_per_env, success_per_env def no_op_and_random_rollout( env: UnderspecifiedEnv, env_params: EnvParams, rng: chex.PRNGKey, init_obs: Observation, init_env_state: EnvState, num_envs: int, max_episode_length: int, ): returns_noop, lens_noop, success_noop = no_op_rollout( env, env_params, rng, init_obs, init_env_state, num_envs, max_episode_length, do_random=False ) returns_random, lens_random, success_random = no_op_rollout( env, env_params, rng, init_obs, init_env_state, num_envs, max_episode_length, do_random=True ) return returns_noop, lens_noop, success_noop, returns_random, lens_random, success_random