diff --git a/tests/imputation/segrnn.py b/tests/imputation/segrnn.py new file mode 100644 index 00000000..17b877ad --- /dev/null +++ b/tests/imputation/segrnn.py @@ -0,0 +1,117 @@ +""" +Test cases for SegRNN imputation model. +""" + +# Created by Wenjie Du +# License: BSD-3-Clause + + +import os.path +import unittest + +import numpy as np +import pytest + +from pypots.imputation import SegRNN +from pypots.optim import Adam +from pypots.utils.logging import logger +from pypots.utils.metrics import calc_mse +from tests.global_test_config import ( + DATA, + EPOCHS, + DEVICE, + TRAIN_SET, + VAL_SET, + TEST_SET, + GENERAL_H5_TRAIN_SET_PATH, + GENERAL_H5_VAL_SET_PATH, + GENERAL_H5_TEST_SET_PATH, + RESULT_SAVING_DIR_FOR_IMPUTATION, + check_tb_and_model_checkpoints_existence, +) + + +class TestSegRNN(unittest.TestCase): + logger.info("Running tests for an imputation model SegRNN...") + + # set the log and model saving path + saving_path = os.path.join(RESULT_SAVING_DIR_FOR_IMPUTATION, "SegRNN") + model_save_name = "saved_segrnn_model.pypots" + + # initialize an Adam optimizer + optimizer = Adam(lr=0.001, weight_decay=1e-5) + + # initialize a SegRNN model + segrnn = SegRNN( + DATA["n_steps"], + DATA["n_features"], + seg_len=4, + dropout=0, + epochs=EPOCHS, + saving_path=saving_path, + optimizer=optimizer, + device=DEVICE, + ) + + @pytest.mark.xdist_group(name="imputation-segrnn") + def test_0_fit(self): + self.segrnn.fit(TRAIN_SET, VAL_SET) + + @pytest.mark.xdist_group(name="imputation-segrnn") + def test_1_impute(self): + imputation_results = self.segrnn.predict(TEST_SET) + assert not np.isnan( + imputation_results["imputation"] + ).any(), "Output still has missing values after running impute()." + + test_MSE = calc_mse( + imputation_results["imputation"], + DATA["test_X_ori"], + DATA["test_X_indicating_mask"], + ) + logger.info(f"SegRNN test_MSE: {test_MSE}") + + @pytest.mark.xdist_group(name="imputation-segrnn") + def test_2_parameters(self): + assert hasattr(self.segrnn, "model") and self.segrnn.model is not None + + assert hasattr(self.segrnn, "optimizer") and self.segrnn.optimizer is not None + + assert hasattr(self.segrnn, "best_loss") + self.assertNotEqual(self.segrnn.best_loss, float("inf")) + + assert hasattr(self.segrnn, "best_model_dict") and self.segrnn.best_model_dict is not None + + @pytest.mark.xdist_group(name="imputation-segrnn") + def test_3_saving_path(self): + # whether the root saving dir exists, which should be created by save_log_into_tb_file + assert os.path.exists(self.saving_path), f"file {self.saving_path} does not exist" + + # check if the tensorboard file and model checkpoints exist + check_tb_and_model_checkpoints_existence(self.segrnn) + + # save the trained model into file, and check if the path exists + saved_model_path = os.path.join(self.saving_path, self.model_save_name) + self.segrnn.save(saved_model_path) + + # test loading the saved model, not necessary, but need to test + self.segrnn.load(saved_model_path) + + @pytest.mark.xdist_group(name="imputation-segrnn") + def test_4_lazy_loading(self): + self.segrnn.fit(GENERAL_H5_TRAIN_SET_PATH, GENERAL_H5_VAL_SET_PATH) + imputation_results = self.segrnn.predict(GENERAL_H5_TEST_SET_PATH) + assert not np.isnan( + imputation_results["imputation"] + ).any(), "Output still has missing values after running impute()." + + test_MSE = calc_mse( + imputation_results["imputation"], + DATA["test_X_ori"], + DATA["test_X_indicating_mask"], + ) + logger.info(f"Lazy-loading SegRNN test_MSE: {test_MSE}") + + +if __name__ == "__main__": + unittest.main()