detectron2/tests/test_checkpoint.py

# Copyright (c) Facebook, Inc. and its affiliates.
import os
import tempfile
import unittest
from collections import OrderedDict
import torch
from iopath.common.file_io import PathHandler, PathManager
from torch import nn

from detectron2.checkpoint import DetectionCheckpointer
from detectron2.checkpoint.c2_model_loading import (
    _longest_common_prefix_str,
    align_and_update_state_dicts,
)
from detectron2.utils.logger import setup_logger


class TestCheckpointer(unittest.TestCase):
    def setUp(self):
        setup_logger()

    def create_complex_model(self):
        m = nn.Module()
        m.block1 = nn.Module()
        m.block1.layer1 = nn.Linear(2, 3)
        m.layer2 = nn.Linear(3, 2)
        m.res = nn.Module()
        m.res.layer2 = nn.Linear(3, 2)

        state_dict = OrderedDict()
        state_dict["layer1.weight"] = torch.rand(3, 2)
        state_dict["layer1.bias"] = torch.rand(3)
        state_dict["layer2.weight"] = torch.rand(2, 3)
        state_dict["layer2.bias"] = torch.rand(2)
        state_dict["res.layer2.weight"] = torch.rand(2, 3)
        state_dict["res.layer2.bias"] = torch.rand(2)
        return m, state_dict

    def test_complex_model_loaded(self):
        for add_data_parallel in [False, True]:
            model, state_dict = self.create_complex_model()
            if add_data_parallel:
                model = nn.DataParallel(model)
            model_sd = model.state_dict()

            sd_to_load = align_and_update_state_dicts(model_sd, state_dict)
            model.load_state_dict(sd_to_load)
            for loaded, stored in zip(model_sd.values(), state_dict.values()):
                # different tensor references
                self.assertFalse(id(loaded) == id(stored))
                # same content
                self.assertTrue(loaded.to(stored).equal(stored))

    def test_load_with_matching_heuristics(self):
        with tempfile.TemporaryDirectory(prefix="detectron2_test") as d:
            model, state_dict = self.create_complex_model()
            torch.save({"model": state_dict}, os.path.join(d, "checkpoint.pth"))
            checkpointer = DetectionCheckpointer(model, save_dir=d)

            with torch.no_grad():
                # use a different weight from the `state_dict`, since torch.rand is less than 1
                model.block1.layer1.weight.fill_(1)

            # load checkpoint without matching_heuristics
            checkpointer.load(os.path.join(d, "checkpoint.pth"))
            self.assertTrue(model.block1.layer1.weight.equal(torch.ones(3, 2)))

            # load checkpoint with matching_heuristics
            checkpointer.load(os.path.join(d, "checkpoint.pth?matching_heuristics=True"))
            self.assertFalse(model.block1.layer1.weight.equal(torch.ones(3, 2)))

    def test_custom_path_manager_handler(self):
        with tempfile.TemporaryDirectory(prefix="detectron2_test") as d:

            class CustomPathManagerHandler(PathHandler):
                PREFIX = "detectron2_test://"

                def _get_supported_prefixes(self):
                    return [self.PREFIX]

                def _get_local_path(self, path, **kwargs):
                    name = path[len(self.PREFIX) :]
                    return os.path.join(d, name)

                def _open(self, path, mode="r", **kwargs):
                    return open(self._get_local_path(path), mode, **kwargs)

            pathmgr = PathManager()
            pathmgr.register_handler(CustomPathManagerHandler())

            model, state_dict = self.create_complex_model()
            torch.save({"model": state_dict}, os.path.join(d, "checkpoint.pth"))
            checkpointer = DetectionCheckpointer(model, save_dir=d)
            checkpointer.path_manager = pathmgr
            checkpointer.load("detectron2_test://checkpoint.pth")
            checkpointer.load("detectron2_test://checkpoint.pth?matching_heuristics=True")

    def test_lcp(self):
        self.assertEqual(_longest_common_prefix_str(["class", "dlaps_model"]), "")
        self.assertEqual(_longest_common_prefix_str(["classA", "classB"]), "class")
        self.assertEqual(_longest_common_prefix_str(["classA", "classB", "clab"]), "cla")


if __name__ == "__main__":
    unittest.main()