model/diffusion/run.py

# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

import argparse
import logging
import os
import random
import re
from glob import glob
from typing import Any, Mapping, TypeVar

import numpy as np
import pytorch_lightning as pl
import torch
import yaml
from hydra.utils import instantiate
from omegaconf import DictConfig, OmegaConf
from pytorch_lightning import Callback
from pytorch_lightning.callbacks import ModelCheckpoint
from pytorch_lightning.cli import SaveConfigCallback
from pytorch_lightning.utilities import rank_zero_only

from ..common.utils.config_utils import get_config
from ..diffusion.config import Config
from ..diffusion.exceptions import AmbiguousConfig
from ..diffusion.lightning_module import DiffusionLightningModule

T = TypeVar("T")

# logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def maybe_instantiate(instance_or_config: T | Mapping, expected_type=None, **kwargs) -> T:
    """
    If instance_or_config is a mapping with a _target_ field, instantiate it.
    Otherwise, return it as is.
    """
    if isinstance(instance_or_config, Mapping) and "_target_" in instance_or_config:
        instance = instantiate(instance_or_config, **kwargs)
    else:
        instance = instance_or_config
    assert expected_type is None or isinstance(
        instance, expected_type
    ), f"Expected {expected_type}, got {type(instance)}"
    return instance


def _find_latest_checkpoint(dirpath: str) -> str | None:
    """Finds the most recent checkpoint inside `dirpath`."""

    # checkpoint names are like "epoch=0-step=0.ckpt."
    # Find the checkpoint with highest epoch:
    def extract_epoch(ckpt):
        match = re.search(r"epoch=(\d+)", ckpt)
        if match:
            return int(match.group(1))
        return -1

    ckpts = glob(f"{dirpath}/*.ckpt")
    epochs = np.array([extract_epoch(ckpt) for ckpt in ckpts])
    if len(epochs) == 0 or epochs.max() < 0:
        # No checkpoints found.
        return None
    latest_checkpoint = ckpts[epochs.argmax()]
    return latest_checkpoint


class SimpleParser:
    def save(self, config, path, **_):
        with open(path, "w") as f:
            yaml.dump(config, f)


class AddConfigCallback(Callback):
    """Adds a copy of the config to the checkpoint, so that `load_from_checkpoint` can use it to instantiate everything."""

    def __init__(self, config: dict[str, Any]):
        self._config_dict = config

    def on_save_checkpoint(
        self, trainer: "pl.Trainer", pl_module: "pl.LightningModule", checkpoint: dict[str, Any]
    ) -> None:
        checkpoint["config"] = self._config_dict


def main(
    config: Config | DictConfig, save_config: bool = True, seed: int | None = None
) -> tuple[pl.Trainer, pl.LightningModule]:
    """
    Main entry point to train and evaluate a diffusion model.

    save_config: if True, the config will be saved both as a YAML file and in each checkpoint. This doesn't work if the config contains things that can't be `yaml.dump`-ed, so
    if you don't care about saving and loading checkpoints and want to use a config that contains things like `torch.nn.Module`s already instantiated, set this to False.
    """
    if config.checkpoint_path and config.auto_resume:
        raise AmbiguousConfig(
            f"Ambiguous config: you set both a checkpoint path {config.checkpoint_path} and `auto_resume` which means automatically select a checkpoint path to resume from."
        )

    if seed is not None:
        torch.manual_seed(seed)
        torch.cuda.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)
        np.random.seed(seed)
        random.seed(seed)

    trainer: pl.Trainer = maybe_instantiate(config.trainer, pl.Trainer)

    if save_config:
        if isinstance(config, DictConfig):
            config_as_dict = OmegaConf.to_container(config, resolve=True)
            # This callback will save a config.yaml file.
            trainer.callbacks.append(
                SaveConfigCallback(
                    parser=SimpleParser(),
                    config=config_as_dict,
                    overwrite=True if config.auto_resume else False,
                )
            )

            # This callback will add a copy of the config to each checkpoint.
            trainer.callbacks.append(AddConfigCallback(config_as_dict))
        else:
            raise NotImplementedError
    datamodule: pl.LightningDataModule = maybe_instantiate(
        config.data_module, pl.LightningDataModule
    )

    # If checkpoint_path is provided training will be resumed from this point.
    # Beware: the old checkpoint will be deleted when a new one is saved.

    ckpt_path = config.checkpoint_path
    if config.auto_resume:
        # Add an additional checkpointer with a fixed directory path to restore from.
        dirpath = os.path.join(trainer.default_root_dir, "checkpoints")
        trainer.callbacks.append(ModelCheckpoint(dirpath=dirpath))
        ckpt_path = _find_latest_checkpoint(dirpath)
    pl_module: DiffusionLightningModule = maybe_instantiate(
        config.lightning_module, DiffusionLightningModule
    )
    if rank_zero_only.rank == 0 and isinstance(trainer.logger, pl.loggers.WandbLogger):
        # Log the config to wandb so that it shows up in the portal.
        trainer.logger.experiment.config.update(
            {**OmegaConf.to_container(config, resolve=True)},
            allow_val_change=True,
        )
    trainer.fit(
        pl_module,
        datamodule=datamodule,
        ckpt_path=ckpt_path,
    )

    return trainer, pl_module


def cli(argv: list[str] | None) -> None:
    """
    Args:
        argv: list of command-line arguments as strings, or None. If None,
          command-line arguments will be got from sys.argv
    """

    parser = argparse.ArgumentParser(allow_abbrev=False)  # prevent prefix matching issues
    parser.add_argument(
        "--seed",
        type=int,
        help="Random seed to use. If not provided, a random seed will be used.",
    )
    args, argv = parser.parse_known_args(argv)

    # Create config from command-line arguments.
    config = get_config(argv, Config)
    main(config, seed=args.seed)


if __name__ == "__main__":
    cli(argv=None)