gitea_admin/sientia-dataops-model-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1249: Refactor TrainModelParams to use dataclass and add from_dict method for validation, remove no-cache-dir from pip install in quality gate workflow, and rename X_train/X_test to x_train/x_test in TrainModelResult.

Browse Source
This commit is contained in:
Bruno Domingues
2025-10-06 19:11:02 -03:00
parent d8583cdae7
commit 04c94efd98
5 changed files with 153 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
model_manager/utils/models/train_model_params.py
View File

@@ -1,6 +1,8 @@
from dataclasses import dataclass
from typing import Any
@dataclass
class TrainModelParams:
"""
Parameters for machine learning model training.
@@ -10,14 +12,17 @@ class TrainModelParams:
tracking information. All parameters are validated upon initialization to ensure
data integrity and prevent runtime errors.
Use the `from_dict()` class method to create instances from dictionaries with
automatic validation of all fields.
Attributes:
variable_columns (List[str]): List of variable column names to use as features.
variable_columns (list[str]): List of variable column names to use as features.
lag_train (int): Number of lags to apply during training phase.
lag_val (int): Number of lags to apply during validation phase.
target_variable (str): Name of the target variable to predict.
rem_static_win (bool): Whether to remove static windows from data.
low_lim (Dict[str, float]): Dictionary of lower limits for each variable.
upp_lim (Dict[str, float]): Dictionary of upper limits for each variable.
low_lim (dict[str, float]): Dictionary of lower limits for each variable.
upp_lim (dict[str, float]): Dictionary of upper limits for each variable.
window (int): Window size for rolling operations.
use_scaler (bool): Whether to use a scaler for data normalization.
include_ar (bool): Whether to include autoregressive variables.
@@ -33,90 +38,92 @@ class TrainModelParams:
removed_intervals (list): List of time intervals to remove from the data.
"""
def __init__(
self,
variable_columns: Any | None,
lag_train: Any | None,
lag_val: Any | None,
target_variable: Any | None,
rem_static_win: Any | None,
low_lim: Any | None,
upp_lim: Any | None,
window: Any | None,
use_scaler: Any | None,
include_ar: Any | None,
bucket_name: Any | None,
file_name: Any | None,
line_separator: Any | None,
decimal_separator: Any | None,
train_size: Any | None,
shuffle: Any | None,
experiment_run_id: Any | None,
experiment_name: Any | None,
experiment_description: Any | None,
removed_intervals: Any | None,
):
"""
Initialize the TrainModelParams object and validate all input parameters.
variable_columns: list[str]
lag_train: int
lag_val: int
target_variable: str
rem_static_win: bool
low_lim: dict[str, float]
upp_lim: dict[str, float]
window: int
use_scaler: bool
include_ar: bool
bucket_name: str
file_name: str
line_separator: str
decimal_separator: str
train_size: int
shuffle: bool
experiment_run_id: int
experiment_name: str
experiment_description: str
removed_intervals: list
This constructor validates that all required parameters are provided and
have the correct types. It raises descriptive errors if validation fails,
helping to catch configuration issues early in the pipeline.
@classmethod
def from_dict(cls, data: dict[str, Any]) -> 'TrainModelParams':
"""
Create TrainModelParams from dictionary with validation.
This factory method creates a TrainModelParams instance from a dictionary,
applying validation to ensure all required fields are present and have
the correct types. This is the recommended way to create instances from
workflow input data.
Args:
variable_columns (list | None): List of variable column names.
lag_train (int | None): Number of lags to apply during training.
lag_val (int | None): Number of lags to apply during validation.
target_variable (str | None): Name of the target variable.
rem_static_win (bool | None): Whether to remove static windows.
low_lim (dict | None): Dictionary of lower limits for variables.
upp_lim (dict | None): Dictionary of upper limits for variables.
window (int | None): Window size for rolling operations.
use_scaler (bool | None): Whether to use a scaler for normalization.
include_ar (bool | None): Whether to include autoregressive variables.
bucket_name (str | None): Name of the MinIO bucket.
file_name (str | None): Name of the file in the MinIO bucket.
line_separator (str | None): Line separator used in the file.
decimal_separator (str | None): Decimal separator used in the file.
train_size (int | None): Percentage of data to use for training.
shuffle (bool | None): Whether to shuffle the data during splitting.
experiment_run_id (int | None): ID of the experiment run.
experiment_name (str | None): Name of the experiment.
experiment_description (str | None): Description of the experiment.
removed_intervals (list | None): List of intervals to remove from the data.
data: Dictionary containing training parameters with keys matching
the attribute names (variable_columns, lag_train, etc.)
Returns:
TrainModelParams: Validated instance with all fields populated
Raises:
ValueError: If any required parameter is None.
TypeError: If any parameter is not of the expected type.
ValueError: If any required field is missing or None
TypeError: If any field has an incorrect type
KeyError: If any required key is missing from the dictionary
Example:
>>> input_data = {
... 'variable_columns': ['var1', 'var2'],
... 'lag_train': 5,
... # ... other fields
... }
>>> params = TrainModelParams.from_dict(input_data)
"""
self.variable_columns: list[str] = self._check_none(
variable_columns, list, 'variable_columns'
)
self.lag_train: int = self._check_none(lag_train, int, 'lag_train')
self.lag_val: int = self._check_none(lag_val, int, 'lag_val')
self.target_variable: str = self._check_none(target_variable, str, 'target_variable')
self.rem_static_win: bool = self._check_none(rem_static_win, bool, 'rem_static_win')
self.low_lim: dict[str, float] = self._check_none(low_lim, dict, 'low_lim')
self.upp_lim: dict[str, float] = self._check_none(upp_lim, dict, 'upp_lim')
self.window: int = self._check_none(window, int, 'window')
self.use_scaler: bool = self._check_none(use_scaler, bool, 'use_scaler')
self.include_ar: bool = self._check_none(include_ar, bool, 'include_ar')
self.bucket_name: str = self._check_none(bucket_name, str, 'bucket_name')
self.file_name: str = self._check_none(file_name, str, 'file_name')
self.line_separator: str = self._check_none(line_separator, str, 'line_separator')
self.decimal_separator: str = self._check_none(decimal_separator, str, 'decimal_separator')
self.train_size: int = self._check_none(train_size, int, 'train_size')
self.shuffle: bool = self._check_none(shuffle, bool, 'shuffle')
self.experiment_run_id: int = self._check_none(experiment_run_id, int, 'experiment_run_id')
self.experiment_name: str = self._check_none(experiment_name, str, 'experiment_name')
self.experiment_description: str = self._check_none(
experiment_description, str, 'experiment_description'
)
self.removed_intervals: list = self._check_type(
removed_intervals, list, 'removed_intervals'
return cls(
variable_columns=cls._check_none(
data.get('variable_columns'), list, 'variable_columns'
),
lag_train=cls._check_none(data.get('lag_train'), int, 'lag_train'),
lag_val=cls._check_none(data.get('lag_val'), int, 'lag_val'),
target_variable=cls._check_none(data.get('target_variable'), str, 'target_variable'),
rem_static_win=cls._check_none(data.get('rem_static_win'), bool, 'rem_static_win'),
low_lim=cls._check_none(data.get('low_lim'), dict, 'low_lim'),
upp_lim=cls._check_none(data.get('upp_lim'), dict, 'upp_lim'),
window=cls._check_none(data.get('window'), int, 'window'),
use_scaler=cls._check_none(data.get('use_scaler'), bool, 'use_scaler'),
include_ar=cls._check_none(data.get('include_ar'), bool, 'include_ar'),
bucket_name=cls._check_none(data.get('bucket_name'), str, 'bucket_name'),
file_name=cls._check_none(data.get('file_name'), str, 'file_name'),
line_separator=cls._check_none(data.get('line_separator'), str, 'line_separator'),
decimal_separator=cls._check_none(
data.get('decimal_separator'), str, 'decimal_separator'
),
train_size=cls._check_none(data.get('train_size'), int, 'train_size'),
shuffle=cls._check_none(data.get('shuffle'), bool, 'shuffle'),
experiment_run_id=cls._check_none(
data.get('experiment_run_id'), int, 'experiment_run_id'
),
experiment_name=cls._check_none(data.get('experiment_name'), str, 'experiment_name'),
experiment_description=cls._check_none(
data.get('experiment_description'), str, 'experiment_description'
),
removed_intervals=cls._check_type(
data.get('removed_intervals'), list, 'removed_intervals'
),
)
def _check_none(self, value: Any | None, expected_type: type, field_name: str):
@staticmethod
def _check_none(value: Any | None, expected_type: type, field_name: str) -> Any:
"""
Validate that a value is not None and check its type.
@@ -139,9 +146,10 @@ class TrainModelParams:
error = f'{field_name} is required and cannot be None.'
raise ValueError(error)
return self._check_type(value, expected_type, field_name)
return TrainModelParams._check_type(value, expected_type, field_name)
def _check_type(self, value: Any | None, expected_type: type, field_name: str):
@staticmethod
def _check_type(value: Any | None, expected_type: type, field_name: str) -> Any:
"""
Validate that a value matches the expected type.

26
model_manager/utils/models/train_model_result.py
View File

@@ -19,27 +19,27 @@ class TrainModelResult:
Attributes:
params (TrainModelParams): The parameters used to train the model.
process_data (DataPreprocessor): The data preprocessor object used to process the input data.
X_train (pd.DataFrame): The training dataset features.
X_test (pd.DataFrame): The testing dataset features.
x_train (pd.DataFrame): The training dataset features.
x_test (pd.DataFrame): The testing dataset features.
y_train (pd.DataFrame): The training dataset target values.
y_test (pd.DataFrame): The testing dataset target values.
regr (LinearRegressionModel): The trained linear regression model.
scaler_dict (dict): A dictionary containing the scalers used to scale the features and target values.
y_pred (Optional[pd.Series]): The predicted target values for the testing dataset. Default is None.
mse_val (Optional[float]): The Mean Squared Error (MSE) of the predictions. Default is None.
mae_val (Optional[float]): The Mean Absolute Error (MAE) of the predictions. Default is None.
r2_val (Optional[float]): The R-squared (R²) value of the predictions. Default is None.
run_name (Optional[str]): The name of the MLFlow run. Default is None.
report_path (Optional[str]): The path to the generated HTML report file. Default is None.
train_data_path (Optional[str]): The path to the training dataset CSV file. Default is None.
test_data_path (Optional[str]): The path to the testing dataset CSV file. Default is None.
run_dir (Optional[str]): The path to the run directory containing all artifacts. Default is None.
y_pred (pd.Series | None): The predicted target values for the testing dataset. Default is None.
mse_val (float | None): The Mean Squared Error (MSE) of the predictions. Default is None.
mae_val (float | None): The Mean Absolute Error (MAE) of the predictions. Default is None.
r2_val (float | None): The R-squared (R²) value of the predictions. Default is None.
run_name (str | None): The name of the MLFlow run. Default is None.
report_path (str | None): The path to the generated HTML report file. Default is None.
train_data_path (str | None): The path to the training dataset CSV file. Default is None.
test_data_path (str | None): The path to the testing dataset CSV file. Default is None.
run_dir (str | None): The path to the run directory containing all artifacts. Default is None.
"""
params: TrainModelParams
process_data: DataPreprocessor
X_train: pd.DataFrame
X_test: pd.DataFrame
x_train: pd.DataFrame
x_test: pd.DataFrame
y_train: pd.DataFrame
y_test: pd.DataFrame
regr: LinearRegressionModel