Skip to content

[ENH] Implement model loading functionality for InceptionTimeRegressor #2726

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 57 additions & 0 deletions aeon/regression/deep_learning/_inception_time.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
57 changes: 57 additions & 0 deletions aeon/regression/deep_learning/_lite_time.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
103 changes: 103 additions & 0 deletions aeon/regression/deep_learning/tests/test_load_model.py
Original file line number Diff line number Diff line change
@@ -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)