train_diffusion.py

"""
Code modified from Diffusion policy: https://github.com/real-stanford/diffusion_policy
--------------------------------------------------------------------------------------
Assumptions made by the code
- The guidance policy is spec first followed by the motion model
- Joint state action has the word 'js' or 'joint' in it's key for the normalizer to deal with it
- Gripper has the work 'gripper' in the key for the normalizer to ignore it
- Action when joint position starts with it followed by the gripper open (last)
"""

import os
import copy
import hydra
import torch
from omegaconf import OmegaConf
import pathlib
from torch.utils.data import DataLoader, Subset
import numpy as np
import wandb
import tqdm
from models.diffusion.diffusion_policy import DiffusionPolicy
from utils.train_utils import BaseTrainer
from utils.misc import (
    get_env_obs_dim,
    get_rob_obs_dim,
    get_action_dim,
    JsonLogger,
    set_seed,
    get_model_ckpt,
    get_4n_dim,
    CKPT_DIR,
)
from diffusers.training_utils import EMAModel
from models.tools.pytorch_util import dict_apply, optimizer_to
from utils.checkpoint_util import TopKCheckpointManager
from models.tools.lr_scheduler import get_scheduler


class TrainDiffusion(BaseTrainer):
    include_keys = ["global_step", "epoch"]

    def __init__(self, cfg: OmegaConf, output_dir=None):
        super().__init__(cfg, output_dir=output_dir)

        # set seed
        seed = cfg.training.seed
        set_seed(seed)

        # configure model
        self.model: DiffusionPolicy = hydra.utils.instantiate(cfg.policy)
        self.ema_model: DiffusionPolicy = None
        if cfg.training.use_ema:
            self.ema_model = copy.deepcopy(self.model)

        # configure motion model
        self.mm_policy = None
        self.mm = cfg.mm
        self.loss_func = self.model.compute_loss

        # configure training state
        self.configure_optimizer()

        self.global_step = 0
        self.epoch = 0

    def configure_optimizer(self):
        pytorch_total_params = sum(
            p.numel() for p in self.model.parameters() if p.requires_grad
        )
        print(f"=== Total trainable parameters: {pytorch_total_params} ===")
        self.optimizer = hydra.utils.instantiate(
            self.cfg.optimizer, params=self.model.parameters()
        )

    def configure_training_type(self):
        cfg = copy.deepcopy(self.cfg)
        if "mm_bb" in cfg and cfg.mm_bb is not None:
            assert not self.mm, "Cannot have a motion model backbone for an mm"
            _, ckpt_path = get_model_ckpt(
                directory=CKPT_DIR,
                ckpt_name=cfg.mm_bb,
                ckpt_tag=["latest.ckpt"],
                # directory=CKPT_DIR, ckpt_name=cfg.mm_bb, ckpt_tag=["min-val-loss", ".ckpt"]
            )
            print(f"=>=>=> Loading motion model from {ckpt_path}")
            mm_trd = TrainDiffusion.create_from_checkpoint(ckpt_path, strict=False)
            try:
                self.mm_policy = mm_trd.ema_model
            except:
                self.mm_policy = mm_trd.model
            self.mm_policy.eval()
            self.mm_policy.to(torch.device(cfg.training.device))
            # change loss function to use mm
            self.loss_func = self.model.compute_cond_loss

    def run(self):
        cfg = copy.deepcopy(self.cfg)

        # fine-tuning for visual adaptation
        if cfg.load_from is not None:
            _, lastest_ckpt_path = get_model_ckpt(
                directory=CKPT_DIR,
                ckpt_name=cfg.load_from[0],
                ckpt_tag=[*cfg.load_from[1:], "latest", ".ckpt"],
            )
            if pathlib.Path(lastest_ckpt_path).is_file():
                print(f"+++++ Resuming from checkpoint {lastest_ckpt_path}")
                self.load_checkpoint(
                    path=lastest_ckpt_path, exclude_keys=["optimizer"], include_keys=[]
                )

        # configure motion model
        self.configure_training_type()

        # configure datasets
        train_dataset, val_dataset = hydra.utils.call(cfg.datasets)
        # define dataloaders
        print(f"Training with {train_dataset.n_demos} demonstrations! ...")
        normalizer = train_dataset.get_normalizer()
        train_dataloader = DataLoader(train_dataset, **cfg.dataloader)
        val_dataloader = DataLoader(val_dataset, **cfg.val_dataloader)

        self.model.set_normalizer(normalizer)
        if cfg.training.use_ema:
            self.ema_model.set_normalizer(normalizer)

        # configure lr scheduler
        lr_scheduler = get_scheduler(
            cfg.training.lr_scheduler,
            optimizer=self.optimizer,
            num_warmup_steps=cfg.training.lr_warmup_steps,
            num_training_steps=(len(train_dataloader) * cfg.training.num_epochs)
            // cfg.training.gradient_accumulate_every,
            # pytorch assumes stepping LRScheduler every epoch
            # however huggingface diffusers steps it every batch
            last_epoch=self.global_step - 1,
        )

        # configure ema
        ema: EMAModel = None
        if cfg.training.use_ema:
            ema = hydra.utils.instantiate(cfg.ema, model=self.ema_model)

        # configure logging
        wandb_run = wandb.init(
            dir=str(self.output_dir),
            config=OmegaConf.to_container(cfg, resolve=True),
            **cfg.logging,
        )
        wandb.config.update({"output_dir": self.output_dir}, allow_val_change=True)

        # configure checkpoint
        topk_manager = TopKCheckpointManager(
            save_dir=os.path.join(self.output_dir, "checkpoints"), **cfg.checkpoint.topk
        )

        # device transfer
        device = torch.device(cfg.training.device)
        self.model.to(device)
        if self.ema_model is not None:
            self.ema_model.to(device)
        optimizer_to(self.optimizer, device)

        # training loop
        log_path = os.path.join(self.output_dir, "logs.json.txt")
        with JsonLogger(log_path) as json_logger:
            for local_epoch_idx in range(cfg.training.num_epochs):
                # save batch for sampling
                test_sampling_batch = None
                step_log = dict()

                # ========= train for this epoch ==========
                train_losses = list()
                with tqdm.tqdm(
                    train_dataloader,
                    desc=f"Training epoch {self.epoch}",
                    leave=False,
                    mininterval=cfg.training.tqdm_interval_sec,
                ) as tepoch:
                    for batch_idx, batch in enumerate(tepoch):
                        # device transfer
                        batch = dict_apply(
                            batch, lambda x: x.to(device, non_blocking=True)
                        )
                        # compute loss
                        raw_loss = self.loss_func(batch, mm_policy=self.mm_policy)
                        loss = raw_loss / cfg.training.gradient_accumulate_every
                        loss.backward()

                        # step optimizer
                        if (
                            self.global_step % cfg.training.gradient_accumulate_every
                            == 0
                        ):
                            self.optimizer.step()
                            self.optimizer.zero_grad()
                            lr_scheduler.step()

                        # update ema
                        if cfg.training.use_ema:
                            ema.step(self.model)

                        # logging
                        raw_loss_cpu = raw_loss.item()
                        tepoch.set_postfix(loss=raw_loss_cpu, refresh=False)
                        train_losses.append(raw_loss_cpu)
                        step_log = {
                            "train_loss": raw_loss_cpu,
                            "global_step": self.global_step,
                            "epoch": self.epoch,
                            "lr": lr_scheduler.get_last_lr()[0],
                        }

                        is_last_batch = batch_idx == (len(train_dataloader) - 1)
                        if not is_last_batch:
                            # log of last step is combined with validation and rollout
                            wandb_run.log(step_log, step=self.global_step)
                            json_logger.log(step_log)
                            self.global_step += 1

                        if (cfg.training.max_train_steps is not None) and batch_idx >= (
                            cfg.training.max_train_steps - 1
                        ):
                            break

                # at the end of each epoch
                # replace train_loss with epoch average
                train_loss = np.mean(train_losses)
                step_log["train_loss"] = train_loss

                # ========= eval for this epoch ==========
                policy = self.model
                if cfg.training.use_ema:
                    policy = self.ema_model
                policy.eval()

                # run validation
                if (self.epoch % cfg.training.val_every) == 0:
                    with torch.no_grad():
                        val_losses = list()
                        with tqdm.tqdm(
                            val_dataloader,
                            desc=f"Validation epoch {self.epoch}",
                            leave=False,
                            mininterval=cfg.training.tqdm_interval_sec,
                        ) as tepoch:
                            for batch_idx, batch in enumerate(tepoch):
                                batch = dict_apply(
                                    batch, lambda x: x.to(device, non_blocking=True)
                                )
                                if test_sampling_batch is None:
                                    test_sampling_batch = batch
                                loss = self.loss_func(batch, mm_policy=self.mm_policy)
                                val_losses.append(loss)
                                if (
                                    cfg.training.max_val_steps is not None
                                ) and batch_idx >= (cfg.training.max_val_steps - 1):
                                    break
                        if len(val_losses) > 0:
                            val_loss = torch.mean(torch.tensor(val_losses)).item()
                            # log epoch average validation loss
                            step_log["val_loss"] = val_loss

                # run diffusion sampling on a test batch
                if (
                    self.epoch % cfg.training.sample_every
                ) == 0 and self.mm_policy is None:
                    with torch.no_grad():
                        # sample trajectory from testing set, and evaluate difference
                        batch = test_sampling_batch  # already on gpu
                        gt_action = batch[self.model.action_key]
                        del batch[self.model.action_key]
                        result = policy.predict_action(obs_dict=batch)
                        pred_action = result["action_pred"]
                        mse = torch.nn.functional.mse_loss(pred_action, gt_action)
                        # log
                        step_log["test_action_mse_error"] = mse.item()
                        # release RAM
                        del batch
                        del gt_action
                        del result
                        del pred_action
                        del mse

                # checkpoint
                if (self.epoch % cfg.training.checkpoint_every) == 0:
                    # checkpointing
                    if cfg.checkpoint.save_last_ckpt:
                        self.save_checkpoint(epoch=self.epoch)
                    if cfg.checkpoint.save_last_snapshot:
                        self.save_snapshot()

                    # sanitize metric names
                    metric_dict = dict()
                    for key, value in step_log.items():
                        new_key = key.replace("/", "_")
                        metric_dict[new_key] = value

                    # We can't copy the last checkpoint here
                    # since save_checkpoint uses threads.
                    # therefore at this point the file might have been empty!
                    topk_ckpt_path = topk_manager.get_ckpt_path(metric_dict)
                    if topk_ckpt_path is not None:
                        self.save_checkpoint(path=topk_ckpt_path)
                # ========= eval end for this epoch ==========
                policy.train()

                # end of epoch
                # log of last step is combined with validation and rollout
                wandb_run.log(step_log, step=self.global_step)
                json_logger.log(step_log)
                self.global_step += 1
                self.epoch += 1


OmegaConf.register_new_resolver("eval", eval, replace=True)
OmegaConf.register_new_resolver("get_env_obs_dim", get_env_obs_dim, replace=True)
OmegaConf.register_new_resolver("get_rob_obs_dim", get_rob_obs_dim, replace=True)
OmegaConf.register_new_resolver("get_action_dim", get_action_dim, replace=True)
OmegaConf.register_new_resolver("get_4n_dim", get_4n_dim, replace=True)
OmegaConf.register_new_resolver("ckpt_dir", lambda x: CKPT_DIR, replace=True)


# datasets=adroit ++datasets.filepath="door_human" ++max_demos=5
# datasets=robo_mimic ++datasets.filepath=lift_low_dim ++logging.mode="disabled" ++max_demos=5
# datasets=rl_bench ++datasets.filepath=open_box ++logging.mode="disabled" ++max_demos=50
@hydra.main(
    version_base=None,
    config_path=str(pathlib.Path(__file__).parent.joinpath("config")),
    config_name=pathlib.Path(__file__).stem,
)
def main(cfg: OmegaConf):
    # resolve immediately so all the ${now:} resolvers
    # will use the same time.
    OmegaConf.resolve(cfg)
    train_diff = TrainDiffusion(cfg)
    train_diff.run()
    wandb.finish()


if __name__ == "__main__":
    main()

# conda activate comp_robotics; CUDA_VISIBLE_DEVICES=0 python train_diffusion.py datasets=robo_mimic ++datasets.filepath=lift_low_dim ++max_demos=50
# conda activate comp_robotics; CUDA_VISIBLE_DEVICES=0 python train_diffusion.py datasets=robo_mimic ++datasets.filepath=can_low_dim ++max_demos=50 ++mm=true
# datasets=pcd_rl_bench ++datasets.filepath=pc_open_fridge ++logging.mode="disabled" ++max_demos=5 ++pcd_only=true models@_global_=pcd_dit