diff --git a/aeon/regression/deep_learning/_inception_time.py b/aeon/regression/deep_learning/_inception_time.py index 96e8a38362..27b454b485 100644 --- a/aeon/regression/deep_learning/_inception_time.py +++ b/aeon/regression/deep_learning/_inception_time.py @@ -369,6 +369,63 @@ def _get_test_params(cls, parameter_set="default"): return [param1] + @classmethod + def load_model(cls, file_path, load_best=True): + """ + Load a saved InceptionTime regressor from file. + + Parameters + ---------- + file_path : str + The path to the directory containing the saved model files + load_best : bool, default=True + Whether to load the best model (if save_best_model was True during training) + or the last model (if save_last_model was True during training) + + Returns + ------- + InceptionTimeRegressor + Loaded regressor + """ + import os + from os.path import exists, join + + import tensorflow as tf + + # Ensure file_path ends with a separator + if not file_path.endswith(os.sep): + file_path = file_path + os.sep + + regressor = cls() + regressor.regressors_ = [] + + # Try to load each regressor model + i = 0 + while True: + if load_best: + model_path = join(file_path, f"best_model{i}.keras") + else: + model_path = join(file_path, f"last_model{i}.keras") + + if not exists(model_path): + break + + rgs = IndividualInceptionRegressor() + rgs.model_ = tf.keras.models.load_model(model_path, compile=False) + rgs.is_fitted = True + regressor.regressors_.append(rgs) + i += 1 + + if len(regressor.regressors_) == 0: + raise FileNotFoundError( + f"No valid model files found in {file_path} with prefix " + f"{'best_model' if load_best else 'last_model'}" + ) + + regressor.n_regressors = len(regressor.regressors_) + regressor.is_fitted = True + return regressor + class IndividualInceptionRegressor(BaseDeepRegressor): """Single Inception regressor. diff --git a/aeon/regression/deep_learning/_lite_time.py b/aeon/regression/deep_learning/_lite_time.py index ffd050f176..0736ae83ef 100644 --- a/aeon/regression/deep_learning/_lite_time.py +++ b/aeon/regression/deep_learning/_lite_time.py @@ -301,6 +301,63 @@ def _get_test_params(cls, parameter_set="default"): return [param1, param2] + @classmethod + def load_model(cls, file_path, load_best=True): + """ + Load a saved LITETime regressor from file. + + Parameters + ---------- + file_path : str + The path to the directory containing the saved model files + load_best : bool, default=True + Whether to load the best model (if save_best_model was True during training) + or the last model (if save_last_model was True during training) + + Returns + ------- + LITETimeRegressor + Loaded regressor + """ + import os + from os.path import exists, join + + import tensorflow as tf + + # Ensure file_path ends with a separator + if not file_path.endswith(os.sep): + file_path = file_path + os.sep + + regressor = cls() + regressor.regressors_ = [] + + # Try to load each regressor model + i = 0 + while True: + if load_best: + model_path = join(file_path, f"best_model{i}.keras") + else: + model_path = join(file_path, f"last_model{i}.keras") + + if not exists(model_path): + break + + rgs = IndividualLITERegressor() + rgs.model_ = tf.keras.models.load_model(model_path, compile=False) + rgs.is_fitted = True + regressor.regressors_.append(rgs) + i += 1 + + if len(regressor.regressors_) == 0: + raise FileNotFoundError( + f"No valid model files found in {file_path} with prefix " + f"{'best_model' if load_best else 'last_model'}" + ) + + regressor.n_regressors = len(regressor.regressors_) + regressor.is_fitted = True + return regressor + class IndividualLITERegressor(BaseDeepRegressor): """Single LITE or LITEMV Regressor. diff --git a/aeon/regression/deep_learning/tests/test_load_model.py b/aeon/regression/deep_learning/tests/test_load_model.py new file mode 100644 index 0000000000..7b96878885 --- /dev/null +++ b/aeon/regression/deep_learning/tests/test_load_model.py @@ -0,0 +1,103 @@ +"""Test load_model functionality for deep learning regression ensemble models.""" + +import os +import tempfile + +import numpy as np +import pytest + +from aeon.regression.deep_learning import InceptionTimeRegressor, LITETimeRegressor +from aeon.testing.data_generation import make_example_3d_numpy +from aeon.utils.validation._dependencies import _check_soft_dependencies + + +@pytest.mark.skipif( + not _check_soft_dependencies("tensorflow", severity="none"), + reason="skip test if required soft dependency not available", +) +def test_inception_time_regressor_load_model(): + """Test loading InceptionTimeRegressor models from files.""" + with tempfile.TemporaryDirectory() as tmp: + # Ensure path ends with a separator + tmp_dir = os.path.join(tmp, "") + + # Generate sample data + X_train, y_train = make_example_3d_numpy( + n_cases=10, n_channels=1, n_timepoints=50, regression_target=True + ) + X_test = X_train.copy() + + # Train model with both best and last model saving + reg = InceptionTimeRegressor( + n_epochs=2, + batch_size=4, + n_regressors=2, + save_best_model=True, + save_last_model=True, + file_path=tmp_dir, + ) + reg.fit(X_train, y_train) + + # Get predictions from original model + y_pred_orig = reg.predict(X_test) + + # Load and test best model + reg_best = InceptionTimeRegressor.load_model(tmp_dir, load_best=True) + y_pred_best = reg_best.predict(X_test) + np.testing.assert_array_almost_equal(y_pred_orig, y_pred_best) + + # Load and test last model + reg_last = InceptionTimeRegressor.load_model(tmp_dir, load_best=False) + # Verify model structure instead of comparing predictions + assert len(reg_last.regressors_) == reg.n_regressors + + # Test error case with invalid path + invalid_path = os.path.join("invalid", "path", "") + with pytest.raises(FileNotFoundError): + InceptionTimeRegressor.load_model(invalid_path) + + +@pytest.mark.skipif( + not _check_soft_dependencies("tensorflow", severity="none"), + reason="skip test if required soft dependency not available", +) +def test_lite_time_regressor_load_model(): + """Test loading LITETimeRegressor models from files.""" + with tempfile.TemporaryDirectory() as tmp: + # Ensure path ends with a separator + tmp_dir = os.path.join(tmp, "") + + # Generate sample data + X_train, y_train = make_example_3d_numpy( + n_cases=10, n_channels=1, n_timepoints=50, regression_target=True + ) + X_test = X_train.copy() + + # Train model with both best and last model saving + reg = LITETimeRegressor( + n_epochs=2, + batch_size=4, + n_regressors=2, + save_best_model=True, + save_last_model=True, + file_path=tmp_dir, + ) + reg.fit(X_train, y_train) + + # Get predictions from original model + y_pred_orig = reg.predict(X_test) + + # Load and test best model + reg_best = LITETimeRegressor.load_model(tmp_dir, load_best=True) + y_pred_best = reg_best.predict(X_test) + np.testing.assert_array_almost_equal(y_pred_orig, y_pred_best) + + # Load and test last model + reg_last = LITETimeRegressor.load_model(tmp_dir, load_best=False) + # Verify model structure instead of comparing predictions + assert len(reg_last.regressors_) == reg.n_regressors + + # Test error case with invalid path + invalid_path = os.path.join("invalid", "path", "") + with pytest.raises(FileNotFoundError): + LITETimeRegressor.load_model(invalid_path)