third_party/SOLD2/sold2/experiment.py

"""
Main file to launch training and testing experiments.
"""

import yaml
import os
import argparse
import numpy as np
import torch

from .config.project_config import Config as cfg
from .train import train_net
from .export import export_predictions, export_homograpy_adaptation


# Pytorch configurations
torch.cuda.empty_cache()
torch.backends.cudnn.benchmark = True


def load_config(config_path):
    """Load configurations from a given yaml file."""
    # Check file exists
    if not os.path.exists(config_path):
        raise ValueError("[Error] The provided config path is not valid.")

    # Load the configuration
    with open(config_path, "r") as f:
        config = yaml.safe_load(f)

    return config


def update_config(path, model_cfg=None, dataset_cfg=None):
    """Update configuration file from the resume path."""
    # Check we need to update or completely override.
    model_cfg = {} if model_cfg is None else model_cfg
    dataset_cfg = {} if dataset_cfg is None else dataset_cfg

    # Load saved configs
    with open(os.path.join(path, "model_cfg.yaml"), "r") as f:
        model_cfg_saved = yaml.safe_load(f)
        model_cfg.update(model_cfg_saved)
    with open(os.path.join(path, "dataset_cfg.yaml"), "r") as f:
        dataset_cfg_saved = yaml.safe_load(f)
        dataset_cfg.update(dataset_cfg_saved)

    # Update the saved yaml file
    if not model_cfg == model_cfg_saved:
        with open(os.path.join(path, "model_cfg.yaml"), "w") as f:
            yaml.dump(model_cfg, f)
    if not dataset_cfg == dataset_cfg_saved:
        with open(os.path.join(path, "dataset_cfg.yaml"), "w") as f:
            yaml.dump(dataset_cfg, f)

    return model_cfg, dataset_cfg


def record_config(model_cfg, dataset_cfg, output_path):
    """Record dataset config to the log path."""
    # Record model config
    with open(os.path.join(output_path, "model_cfg.yaml"), "w") as f:
        yaml.safe_dump(model_cfg, f)

    # Record dataset config
    with open(os.path.join(output_path, "dataset_cfg.yaml"), "w") as f:
        yaml.safe_dump(dataset_cfg, f)


def train(args, dataset_cfg, model_cfg, output_path):
    """Training function."""
    # Update model config from the resume path (only in resume mode)
    if args.resume:
        if os.path.realpath(output_path) != os.path.realpath(args.resume_path):
            record_config(model_cfg, dataset_cfg, output_path)

    # First time, then write the config file to the output path
    else:
        record_config(model_cfg, dataset_cfg, output_path)

    # Launch the training
    train_net(args, dataset_cfg, model_cfg, output_path)


def export(
    args,
    dataset_cfg,
    model_cfg,
    output_path,
    export_dataset_mode=None,
    device=torch.device("cuda"),
):
    """Export function."""
    # Choose between normal predictions export or homography adaptation
    if dataset_cfg.get("homography_adaptation") is not None:
        print("[Info] Export predictions with homography adaptation.")
        export_homograpy_adaptation(
            args, dataset_cfg, model_cfg, output_path, export_dataset_mode, device
        )
    else:
        print("[Info] Export predictions normally.")
        export_predictions(
            args, dataset_cfg, model_cfg, output_path, export_dataset_mode
        )


def main(
    args, dataset_cfg, model_cfg, export_dataset_mode=None, device=torch.device("cuda")
):
    """Main function."""
    # Make the output path
    output_path = os.path.join(cfg.EXP_PATH, args.exp_name)

    if args.mode == "train":
        if not os.path.exists(output_path):
            os.makedirs(output_path)
        print("[Info] Training mode")
        print("\t Output path: %s" % output_path)
        train(args, dataset_cfg, model_cfg, output_path)
    elif args.mode == "export":
        # Different output_path in export mode
        output_path = os.path.join(cfg.export_dataroot, args.exp_name)
        print("[Info] Export mode")
        print("\t Output path: %s" % output_path)
        export(
            args,
            dataset_cfg,
            model_cfg,
            output_path,
            export_dataset_mode,
            device=device,
        )
    else:
        raise ValueError("[Error]: Unknown mode: " + args.mode)


def set_random_seed(seed):
    np.random.seed(seed)
    torch.manual_seed(seed)


if __name__ == "__main__":
    # Parse input arguments
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--mode", type=str, default="train", help="'train' or 'export'."
    )
    parser.add_argument(
        "--dataset_config", type=str, default=None, help="Path to the dataset config."
    )
    parser.add_argument(
        "--model_config", type=str, default=None, help="Path to the model config."
    )
    parser.add_argument("--exp_name", type=str, default="exp", help="Experiment name.")
    parser.add_argument(
        "--resume",
        action="store_true",
        default=False,
        help="Load a previously trained model.",
    )
    parser.add_argument(
        "--pretrained",
        action="store_true",
        default=False,
        help="Start training from a pre-trained model.",
    )
    parser.add_argument(
        "--resume_path", default=None, help="Path from which to resume training."
    )
    parser.add_argument(
        "--pretrained_path", default=None, help="Path to the pre-trained model."
    )
    parser.add_argument(
        "--checkpoint_name", default=None, help="Name of the checkpoint to use."
    )
    parser.add_argument(
        "--export_dataset_mode", default=None, help="'train' or 'test'."
    )
    parser.add_argument(
        "--export_batch_size", default=4, type=int, help="Export batch size."
    )

    args = parser.parse_args()

    # Check if GPU is available
    # Get the model
    if torch.cuda.is_available():
        device = torch.device("cuda")
    else:
        device = torch.device("cpu")

    # Check if dataset config and model config is given.
    if (
        ((args.dataset_config is None) or (args.model_config is None))
        and (not args.resume)
        and (args.mode == "train")
    ):
        raise ValueError(
            "[Error] The dataset config and model config should be given in non-resume mode"
        )

    # If resume, check if the resume path has been given
    if args.resume and (args.resume_path is None):
        raise ValueError("[Error] Missing resume path.")

    # [Training] Load the config file.
    if args.mode == "train" and (not args.resume):
        # Check the pretrained checkpoint_path exists
        if args.pretrained:
            checkpoint_folder = args.resume_path
            checkpoint_path = os.path.join(args.pretrained_path, args.checkpoint_name)
            if not os.path.exists(checkpoint_path):
                raise ValueError("[Error] Missing checkpoint: " + checkpoint_path)
        dataset_cfg = load_config(args.dataset_config)
        model_cfg = load_config(args.model_config)

    # [resume Training, Test, Export] Load the config file.
    elif (args.mode == "train" and args.resume) or (args.mode == "export"):
        # Check checkpoint path exists
        checkpoint_folder = args.resume_path
        checkpoint_path = os.path.join(args.resume_path, args.checkpoint_name)
        if not os.path.exists(checkpoint_path):
            raise ValueError("[Error] Missing checkpoint: " + checkpoint_path)

        # Load model_cfg from checkpoint folder if not provided
        if args.model_config is None:
            print("[Info] No model config provided. Loading from checkpoint folder.")
            model_cfg_path = os.path.join(checkpoint_folder, "model_cfg.yaml")
            if not os.path.exists(model_cfg_path):
                raise ValueError("[Error] Missing model config in checkpoint path.")
            model_cfg = load_config(model_cfg_path)
        else:
            model_cfg = load_config(args.model_config)

        # Load dataset_cfg from checkpoint folder if not provided
        if args.dataset_config is None:
            print("[Info] No dataset config provided. Loading from checkpoint folder.")
            dataset_cfg_path = os.path.join(checkpoint_folder, "dataset_cfg.yaml")
            if not os.path.exists(dataset_cfg_path):
                raise ValueError("[Error] Missing dataset config in checkpoint path.")
            dataset_cfg = load_config(dataset_cfg_path)
        else:
            dataset_cfg = load_config(args.dataset_config)

        # Check the --export_dataset_mode flag
        if (args.mode == "export") and (args.export_dataset_mode is None):
            raise ValueError("[Error] Empty --export_dataset_mode flag.")
    else:
        raise ValueError("[Error] Unknown mode: " + args.mode)

    # Set the random seed
    seed = dataset_cfg.get("random_seed", 0)
    set_random_seed(seed)

    main(
        args,
        dataset_cfg,
        model_cfg,
        export_dataset_mode=args.export_dataset_mode,
        device=device,
    )