diff --git a/encode.sh b/encode.sh new file mode 100755 index 0000000..2beb0d1 --- /dev/null +++ b/encode.sh @@ -0,0 +1,13 @@ +source ./venv/bin/activate + +pip install pathspec +pip install pyyaml + +echo " +.git" >> .gitignore + +python encrypt.py ./ code --ignore .gitignore --chunk-size 100000 + +sed -i '/.git/d' .gitignore + +xdg-open . \ No newline at end of file diff --git a/encrypt.py b/encrypt.py new file mode 100644 index 0000000..6aff252 --- /dev/null +++ b/encrypt.py @@ -0,0 +1,112 @@ +import os +import argparse +from pathspec import PathSpec +import yaml + +''' +Usage: + python .\encrypt.py path_to_dir output_file --ignore ignore_file --chunk-size 100000 +''' + + +def load_ignore_patterns(ignore_file, include_library): + # Ensure the .gitignore file exists + if not os.path.exists(ignore_file): + raise FileNotFoundError(f"Ignore file not found at {ignore_file}") + + # Load and parse the .gitignore patterns + with open(ignore_file, 'r') as file: + patterns = file.readlines() + if not include_library: + patterns.append('**/deploy/library/') + + spec = PathSpec.from_lines('gitwildmatch', patterns) + return spec + + +def is_ignored(file_path, spec): + """Check if a file should be ignored based on the ignore patterns.""" + return spec.match_file(file_path) if spec else False + + +def encode_file_tree_to_yaml(directory, ignore_file, include_library): + """Encode the file tree into a single YAML file.""" + ignore_patterns = load_ignore_patterns( + ignore_file, include_library) if ignore_file else None + file_tree = {} + + for root, dirs, files in os.walk(directory): + # Skip ignored directories + dirs[:] = [d for d in dirs if not is_ignored( + os.path.join(root, d), ignore_patterns)] + + for file in files: + file_path = os.path.join(root, file) + + # Skip ignored files + if is_ignored(file_path, ignore_patterns): + continue + + # Read file content + try: + with open(file_path, 'r', encoding='utf-8') as f: + content = f.read() + except Exception as e: + print(f"Error reading file {file_path}: {e}") + raise + + # Create nested dictionary structure + path_parts = os.path.relpath(file_path, directory).split(os.sep) + current_level = file_tree + + # all except the last part (the file name) + for part in path_parts[:-1]: + current_level = current_level.setdefault(part, {}) + + # Add the file and its content + current_level[path_parts[-1]] = content + return yaml.dump(file_tree, default_flow_style=False) + + +def chunk_and_write_file_tree_to_yaml(yaml_content, output_file, chunk_size=None): + """Chunk the YAML content and write it to the output file.""" + + chunks = [yaml_content] if chunk_size is None else [ + yaml_content[i:i + chunk_size] for i in range(0, len(yaml_content), chunk_size)] + + for i, chunk in enumerate(chunks): + chunk_file = f"{output_file}_{i}.yaml" + # Write the file tree to the output YAML file + with open(chunk_file, 'w', encoding='utf-8') as yaml_file: + yaml_file.write(chunk) + + +def main(): + parser = argparse.ArgumentParser( + description="Encrypts file tree to yaml file") + parser.add_argument("input_directory", help="Directory to encode") + parser.add_argument("output_yaml_file", help="Output YAML file") + parser.add_argument("--ignore", default=None, + help="Path to the ignore file") + parser.add_argument("--chunk-size", type=int, default=None, + help="Chunk size for the output YAML file") + parser.add_argument("--library", type=bool, default=False, + help="Incude the library in the output YAML file") + + # Parse arguments + args = parser.parse_args() + + # Example usage + directory_to_encode = args.input_directory + ignore_file_path = args.ignore + output_yaml_file = args.output_yaml_file + include_library = args.library + + content = encode_file_tree_to_yaml( + directory_to_encode, ignore_file_path, include_library) + chunk_and_write_file_tree_to_yaml( + content, output_yaml_file, args.chunk_size) + + +if __name__ == "__main__": + main() diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 0b4ec9f..2c427e0 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -289,7 +289,7 @@ class OPC(BaseActivity): success = success and local_success self.info( - f"Process completed for OPC server {server_id}: {local_count} of {len(config['prediction_tags'])} prediction tags and {len(config['confidence_tags'])} confidence tags", metadata) + f"Process completed for OPC server {server_id}: {local_count} of {len(config.get('prediction_tags', []))} prediction tags and {len(config.get('confidence_tags', []))} confidence tags", metadata) return self.process_confidence(data, success, metadata) diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 145f248..41c933f 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -75,7 +75,7 @@ def build_opc_config() -> Dict[str, Any]: OPC_CERT_PATH: Client certificate path (fallback, default: None) OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None) OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None) - OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback, default: 120) + OPC_RECONNECTION_INTERVAL: Reconnection interval in seconds (fallback, default: 120) Returns: dict: OPC server configuration dictionary diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 68322f3..07f0fc5 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -10,88 +10,75 @@ requests using the Model Monitoring API functions. By Monitoring we mean the evaluation of the performance of models, the generation of reports. """ -from datetime import datetime +from datetime import datetime, timedelta import traceback import pandas as pd import mlflow -from os import makedirs, path, remove -from sientia.ModelServing import ModelServing +from os import makedirs, path, remove, environ +from sys import path as sys_path +from sientia_do.observability.logger import Logger +import lzma +import gzip +import pickle + +ARTIFACTS_PATH = "./tmp/artifacts" class MLFlowRepository(): - def __init__(self, host, username, password, logger): + def __init__(self, host: str, username: str, password: str, logger: Logger): - self.model_serving = ModelServing(tracking_uri=host, - username=username, password=password, - logger=logger) + # set tracking uri + mlflow.set_tracking_uri(host) - def transform(self, model_name: str, data: pd.DataFrame, model_retention: int): + environ["MLFLOW_TRACKING_USERNAME"] = username + environ["MLFLOW_TRACKING_PASSWORD"] = password + # Create an MLflow client + self.client = mlflow.tracking.MlflowClient() + + self.model_cache = {} + + """ + Functions related to get model registry parameters + """ + + def get_model_uri(self, run_id: str, prediction: bool = True): """ - Transform data using a model. + Get the model URI based on the run_id. - Parameters: - - model_name (str): The name of the model to use for transformation. - - data (pandas.DataFrame): The data to transform. - - model_retention (int): The number of minutes to keep the model. + Args: + run_id (str): The run_id of the model. Returns: - - dict: A dictionary containing the transformed data. + str: The model URI. """ + run_info = mlflow.get_run(run_id) + if prediction: + model_uri = run_info.info.artifact_uri + "/prediction_model" + else: + model_uri = run_info.info.artifact_uri + "/data_model" + return model_uri - try: - - return { - 'success': True, - 'content': self.model_serving.get_cached_transform( - model_name, data, model_retention).to_dict() - } - - except Exception as e: - return { - 'success': False, - 'content': { - 'message': str(e), - 'traceback': traceback.format_exc() - } - } - - def predict(self, model_name: str, data: pd.DataFrame, model_retention: int): + def get_model_run_id(self, model_name: str, stage: str = "Production"): """ - Predict data using a model. + Get the run_id of a model based on its name and stage. - Parameters: - - model_name (str): The name of the model to use for prediction. - - data (pandas.DataFrame): The data to predict. - - model_retention (int): The number of minutes to keep the model. + Args: + model_name (str): The name of the model. + stage (str): The stage of the model. Returns: - - dict: A dictionary containing the predicted data. + str: The run_id of the model. """ - try: - - input_index = data.index - start_time = datetime.now() - data = self.model_serving.get_cached_predict( - model_name, data, model_retention) - - end_time = datetime.now() - data = pd.DataFrame(data, columns=['prediction']) - data.index = input_index - data['response_time'] = (end_time - start_time).total_seconds() - - return { - 'success': True, - 'content': data.to_dict() - } - - except Exception as e: - return { - 'success': False, - 'content': { - 'message': str(e), - 'traceback': traceback.format_exc() - } - } + latest_versions = self.client.get_latest_versions( + name=model_name, stages=[stage] + ) + if not latest_versions: + raise mlflow.exceptions.MlflowException( + f"Model '{model_name}' in stage '{stage}' not found in the Model Registry." + ) + else: + run_id = latest_versions[0].source.split("/") + return run_id[2] def get_experiment_by_run_id(self, run_id: str) -> dict: # Get the run information using the run_id @@ -124,6 +111,491 @@ class MLFlowRepository(): next_run_number = len(runs) + 1 return f"{model_name}-{next_run_number}" + def get_experiment(self, experiment_name: str) -> int: + """ + Retrieve MLFlow experiment ID by experiment name. + + This method searches for an MLFlow experiment by name and + returns its unique identifier. It provides error handling + for non-existent experiments. + + Args: + experiment_name (str): Name of the MLFlow experiment + + Returns: + int: MLFlow experiment ID + + Raises: + ValueError: If the experiment name is not found + """ + experiment = mlflow.get_experiment_by_name(experiment_name) + + if experiment is None: + raise ValueError(f'Experiment {experiment_name} not found') + + return int(experiment.experiment_id) + + def get_experiment_last_run(self, experiment_id: int) -> str: + """ + Retrieve the most recent retraining run ID for an experiment. + + This method searches for the latest run in an MLFlow experiment + that has been marked as a retraining run. It filters runs by + the 'retrain' parameter and orders them by completion time. + + Args: + experiment_id (int): MLFlow experiment ID + + Returns: + str: MLFlow run ID of the most recent retraining run + + Raises: + ValueError: If runs data is not in expected DataFrame format + """ + runs = mlflow.search_runs( + experiment_ids=[experiment_id], + filter_string="", # Sem filtro no MLflow ainda + output_format="pandas" + ) + + if not isinstance(runs, pd.DataFrame): + raise ValueError('Runs is not a pandas DataFrame') + + # Filtrar apenas as runs onde params.retrain == True + filtered_runs = runs[runs["params.retrain"] == 'True'] + + # Converter a coluna 'end_time' para datetime + filtered_runs['end_time'] = pd.to_datetime(filtered_runs['end_time']) + + # Ordenar o DataFrame de forma descendente pela coluna 'end_time' + filtered_runs = filtered_runs.sort_values( + by='end_time', ascending=False) + + # Pegar a última run_id do DataFrame filtrado e ordenado + latest_run_id = filtered_runs.iloc[0]['run_id'] + + return latest_run_id + + """ + Functions related to download and load models + """ + + def dowload_artifacts(self, model_name: str, artifact_path: str = "data_model") -> str: + """ + Downloads artifacts from a specific MLFlow run. + """ + run_id = self.get_model_run_id( + model_name=model_name, stage="Production" + ) + output_dir = f"{ARTIFACTS_PATH}/{model_name}" + + if not path.exists(output_dir): + makedirs(output_dir) + + return self.client.download_artifacts( + run_id, + artifact_path, + output_dir + ) + + def load_predict_model(self, model_name: str, flavor: str = 'pyfunc', + artifact_path: str | None = None): + """ + Downloads a predictive model from the MLflow Model Registry. + Args: + model_name (str): The name of the model to download from the registry. + Returns: + mlflow.pyfunc.PyFuncModel: The loaded predictive model. + Notes: + - The model is fetched from the "production" stage of the MLflow Model Registry. + - Warnings during the model loading process are suppressed. + """ + + model_uri = f"models:/{model_name}/production" + + if artifact_path: + self.logger.info( + f"Prediction model {model_name} is not compressed, loading from {artifact_path}") + + model = self.load_model_with_compression( + artifact_path, "prediction") + + else: + if flavor == 'pyfunc': + model = mlflow.pyfunc.load_model(model_uri) + elif flavor == 'sklearn': + model = mlflow.sklearn.load_model(model_uri) + elif flavor == 'pytorch': + model = mlflow.pytorch.load_model(model_uri) + else: + raise ValueError( + "Invalid flavor. Use 'sklearn' or 'pyfunc' or 'pytorch'.") + + return model + + def load_transform_model(self, model_name: str, flavor: str, + artifact_path: str | None = None): + """ + Downloads the latest production version of a specified model. + + This method retrieves the latest production model run ID for the given + model name, constructs the model URI, and loads the model using MLflow. + + Args: + model_name (str): The name of the model to download. + + Returns: + Any: The loaded model object, as returned by `mlflow.sklearn.load_model`. + + Raises: + Exception: If the model run ID or URI cannot be retrieved, or if the + model cannot be loaded. + """ + latest_production_id = self.get_model_run_id( + model_name=model_name, stage="Production" + ) + model_uri = self.get_model_uri( + latest_production_id, prediction=False) + + self.logger.debug( + f"Model {model_name} is at {model_uri} and latest production id is {latest_production_id}") + + if artifact_path: + # Download model artifacts + self.logger.info( + f"Data model {model_name} is not compressed, loading from {artifact_path}") + + model = self.load_model_with_compression( + artifact_path, "transformer") + else: + if flavor == 'sklearn': + model = mlflow.sklearn.load_model(model_uri) + elif flavor == 'pyfunc': + model = mlflow.pyfunc.load_model(model_uri) + elif flavor == 'pytorch': + model = mlflow.pytorch.load_model(model_uri) + else: + raise ValueError( + "Invalid flavor. Use 'sklearn' or 'pyfunc' or 'pytorch'.") + return model + + def load_model_with_compression(self, artifact_path: str, type: str): + """ + Load model from pickle file trying different compression methods. + + Args: + pickle_path (str): Path to the pickle file + + Returns: + Any: Loaded model object + + Raises: + ValueError: If model cannot be loaded with any compression method + """ + + if type == "transformer": + code_path = path.join( + artifact_path, "transformer_pyfunc", "code") + pickle_path = path.join( + artifact_path, "transformer_pyfunc", "artifacts", "training_transformer.pkl") + + elif type == "prediction": + code_path = path.join( + artifact_path, "stacking_model", "code") + pickle_path = path.join( + artifact_path, "stacking_model", "artifacts", "stacking_model.pkl") + + if code_path not in sys_path: + sys_path.insert(0, code_path) + self.logger.info( + f"Added {code_path} to Python path") + + loading_methods = [ + ("lzma", lambda p: lzma.open(p, "rb")), + ("gzip", lambda p: gzip.open(p, "rb")), + ("pickle", lambda p: open(p, "rb")), + ] + + for format_name, open_func in loading_methods: + try: + self.logger.info(f"Trying to load with {format_name}...") + with open_func(pickle_path) as f: + model = pickle.load(f) + self.logger.info( + f"Successfully loaded with {format_name}!") + return model + except (lzma.LZMAError, gzip.BadGzipFile, OSError, pickle.UnpicklingError, ValueError) as e: + self.logger.info( + f"Failed with {format_name}: {e.__class__.__name__}:{e}") + continue + + raise ValueError( + f"Could not load model from {pickle_path} - unknown or corrupted format") + + def download_model(self, model_name: str, model_type: str, flavor: str, + compressed: bool = False) -> dict: + """ + Download model based on type (predict or transform). + + Args: + model_name (str): Name of the model to download + model_type (str): Type of model ('predict' or 'transform') + flavor (str): Model flavor ('sklearn', 'pyfunc', 'pytorch') + compressed (bool): Whether model is compressed + + Returns: + dict: Model configuration with model and artifact paths + """ + if model_type == "predict": + if compressed: + artifact_path = self.dowload_artifacts( + model_name, "prediction_model") + else: + artifact_path = None + model = self.load_predict_model(model_name, flavor, artifact_path) + + elif model_type == "transform": + if compressed: + self.logger.info( + f"Model {model_name} is compressed, downloading artifacts") + + artifact_path = self.dowload_artifacts( + model_name, "data_model") + + self.logger.info( + f"Artifacts downloaded at path {artifact_path}") + else: + artifact_path = None + model = self.load_transform_model( + model_name, flavor, artifact_path) + else: + raise ValueError( + "Invalid model_type. Use 'predict' or 'transform'.") + + return { + "model": model, + "artifact_path": artifact_path + } + + """ + Functions related to cache management of models + """ + + def check_cache_config(self, cache: dict, new_config: dict) -> bool: + """ + Check if cache configuration matches new configuration. + + Args: + cache (dict): Cached model configuration + new_config (dict): New configuration to compare + + Returns: + bool: True if configurations match, False otherwise + """ + old_config = cache['config'] + if old_config != new_config: + return False + return True + + def check_cache_retention(self, cache: dict, retention: int) -> bool: + """ + Check if cache is still valid based on retention time. + + Args: + cache (dict): Cached model data + retention (int): Retention time in minutes + + Returns: + bool: True if cache is still valid, False if expired + """ + current_time = self.now() + cache_time = cache['timestamp'] + if current_time - cache_time >= timedelta(minutes=retention): + return False + return True + + def handle_valid_model(self, model_name: str, model_type: str, + compressed: bool, retention_target: str, + cache: dict) -> dict: + """ + Handle valid cached model by returning appropriate model configuration. + + Args: + model_name (str): Name of the model + model_type (str): Type of model ('predict' or 'transform') + compressed (bool): Whether model is compressed + retention_target (str): Retention target ('model' or 'artifact') + cache (dict): Cached model data + + Returns: + dict: Model configuration with model and artifact path + """ + if self.logger: + self.logger.debug( + f"Model {model_name} is still valid, using cached version") + + # If model is compressed and retention target is artifact, load the model from pkl + if compressed and retention_target == "artifact": + model = self.load_model_with_compression( + cache['target']['artifact_path'], model_type) + return { + 'model': model, + 'artifact_path': cache['target']['artifact_path'] + } + else: + return cache['target'] + + def handle_outdated_model(self, model_name: str, model_key: str) -> dict: + """ + Clean up outdated cached model and its artifacts. + + Args: + model_name (str): Name of the model + model_key (str): Cache key for the model + + Returns: + dict: Empty dictionary (cleanup operation) + """ + if self.logger: + self.logger.debug( + f"Model {model_name} is outdated, downloading a new one") + + del self.model_cache[model_key]['target']['model'] + if path.exists(self.model_cache[model_key]['target']['artifact_path']): + remove(self.model_cache[model_key] + ['target']['artifact_path']) + del self.model_cache[model_key] + + def get_model(self, model_name: str, retention: int, model_type: str, flavor: str, + compressed: bool = False, retention_target: str = "model"): + """ + Get model with caching support based on retention policy. + + Args: + model_name (str): Name of the model to retrieve + retention (int): Cache retention time in minutes (0 = no cache) + model_type (str): Type of model ('predict' or 'transform') + flavor (str): Model flavor ('sklearn', 'pyfunc', 'pytorch') + compressed (bool): Whether model is compressed + retention_target (str): What to cache ('model' or 'artifact') + + Returns: + dict: Model configuration with model and artifact paths + """ + # Retention is 0, download a new model + if retention <= 0: + model_config = self.download_model( + model_name, model_type, flavor, compressed) + return model_config + + model_key = f'{model_name}_{model_type}' + config = { + 'compressed': compressed, + 'retention_target': retention_target + } + + if model_key in self.model_cache: + cache = self.model_cache[model_key] + + # Check if config has changed or is outdated + if self.check_cache_config(cache, config) or self.check_cache_retention(cache, retention): + return self.handle_valid_model( + model_name, model_type, compressed, + retention_target, cache) + else: + # Model is outdated, delete old model files + self.handle_outdated_model(model_name, model_key) + else: + if self.logger: + self.logger.debug( + f"Model {model_name} is not in the cache, downloading a new one") + + # Donwload new model + model_config = self.download_model(model_name, model_type, flavor, + compressed) + # If model is compressed and retention target is artifact, + # dont save the model in the cache + if compressed and retention_target == "artifact": + model_config_to_cache = { + 'artifact_path': model_config['artifact_path'], + 'model': None + } + else: + model_config_to_cache = { + **model_config + } + cache = { + 'target': model_config_to_cache, + 'config': config, + 'timestamp': self.now() + } + + self.model_cache[model_key] = cache + + return model_config + + def get_cached_transform(self, model_name: str, data: pd.DataFrame, retention: int, flavor: str, + compressed: bool = False, retention_target: str = "model", keyword: str = "predict") -> pd.DataFrame: + """ + Get transformed data using cached transform model. + + Args: + model_name (str): Name of the transform model + data (pd.DataFrame): Data to transform + retention (int): Cache retention time in minutes + flavor (str): Model flavor ('sklearn', 'pyfunc', 'pytorch') + compressed (bool): Whether model is compressed + retention_target (str): What to cache ('model' or 'artifact') + keyword (str): Method name to call on model (default: 'predict') + + Returns: + pd.DataFrame: Transformed data + """ + model_config = self.get_model(model_name, retention, + "transform", flavor, compressed, retention_target) + model = model_config['model'] + + # Use getattr to dynamically call the method specified by keyword + method = getattr(model, keyword) + transformed_data = method(data) + + # If model is compressed and retention target is artifact, + # delete the model after the prediction + if compressed and retention_target == "artifact": + del model + + # If retention is 0, delete the artifacts after the prediction + if retention == 0 and model_config['artifact_path'] is not None: + if path.exists(model_config['artifact_path']): + remove(model_config['artifact_path']) + + return transformed_data + + def get_cached_predict(self, model_name: str, data: pd.DataFrame, retention: int, flavor: str, + compressed: bool = False, retention_target: str = "model") -> pd.DataFrame: + """ + Get predictions using cached prediction model. + + Args: + model_name (str): Name of the prediction model + data (pd.DataFrame): Data to make predictions on + retention (int): Cache retention time in minutes + flavor (str): Model flavor ('sklearn', 'pyfunc', 'pytorch') + compressed (bool): Whether model is compressed + retention_target (str): What to cache ('model' or 'artifact') + + Returns: + pd.DataFrame: Model predictions + """ + model_config = self.get_model(model_name, retention, "predict", + flavor, compressed, retention_target) + model = model_config['model'] + return model.predict(data) + + """ + Functions related to model retraining + """ + def create_model_experiment(self, model_name: str, data: pd.DataFrame) -> tuple: """ Create a new MLFlow experiment for model retraining. @@ -148,10 +620,10 @@ class MLFlowRepository(): # load predictor model predictor_uri = f"models:/{model_name}/production" # load transform model - latest_production_id = self.model_serving.get_model_run_id( + latest_production_id = self.get_model_run_id( model_name, stage="Production" ) - transform_uri = self.model_serving.get_model_uri( + transform_uri = self.get_model_uri( latest_production_id, prediction=False ) # load @@ -235,96 +707,6 @@ class MLFlowRepository(): return "Model retrained successfully", experiment - def retrain_model(self, data: pd.DataFrame, model_name: str) -> tuple: - """ - Orchestrate the complete model retraining workflow. - - This method coordinates the entire model retraining process by: - 1. Creating the MLFlow experiment environment - 2. Loading existing production models - 3. Executing the retraining process - 4. Returning comprehensive retraining results - - Args: - data (pd.DataFrame): Training data for model retraining - model_name (str): Name of the MLFlow model to retrain - - Returns: - tuple: (status_message, experiment_name) - - status_message (str): Retraining operation status - - experiment_name (str): MLFlow experiment identifier - """ - prediction_model, data_model, experiment = self.create_model_experiment( - model_name, data) - retrain_result = self.perform_model_retrain( - prediction_model, data_model, experiment, model_name, data) - return retrain_result - - def get_experiment(self, experiment_name: str) -> int: - """ - Retrieve MLFlow experiment ID by experiment name. - - This method searches for an MLFlow experiment by name and - returns its unique identifier. It provides error handling - for non-existent experiments. - - Args: - experiment_name (str): Name of the MLFlow experiment - - Returns: - int: MLFlow experiment ID - - Raises: - ValueError: If the experiment name is not found - """ - experiment = mlflow.get_experiment_by_name(experiment_name) - - if experiment is None: - raise ValueError(f'Experiment {experiment_name} not found') - - return int(experiment.experiment_id) - - def get_experiment_last_run(self, experiment_id: int) -> str: - """ - Retrieve the most recent retraining run ID for an experiment. - - This method searches for the latest run in an MLFlow experiment - that has been marked as a retraining run. It filters runs by - the 'retrain' parameter and orders them by completion time. - - Args: - experiment_id (int): MLFlow experiment ID - - Returns: - str: MLFlow run ID of the most recent retraining run - - Raises: - ValueError: If runs data is not in expected DataFrame format - """ - runs = mlflow.search_runs( - experiment_ids=[experiment_id], - filter_string="", # Sem filtro no MLflow ainda - output_format="pandas" - ) - - if not isinstance(runs, pd.DataFrame): - raise ValueError('Runs is not a pandas DataFrame') - - # Filtrar apenas as runs onde params.retrain == True - filtered_runs = runs[runs["params.retrain"] == 'True'] - - # Converter a coluna 'end_time' para datetime - filtered_runs['end_time'] = pd.to_datetime(filtered_runs['end_time']) - - # Ordenar o DataFrame de forma descendente pela coluna 'end_time' - filtered_runs = filtered_runs.sort_values( - by='end_time', ascending=False) - - # Pegar a última run_id do DataFrame filtrado e ordenado - latest_run_id = filtered_runs.iloc[0]['run_id'] - - return latest_run_id - def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict: """ Update production model with a specific MLFlow run. @@ -382,24 +764,215 @@ class MLFlowRepository(): 'mlflow_run_id': run_id } + """ + Functions that provide the interface to model operations + """ + + def transform(self, model_name: str, data: pd.DataFrame, model_retention: int): + """ + Transform data using a cached transformation model. + + This method provides a high-level interface for data transformation operations + using MLFlow models. It handles model caching, error management, and data + format conversion automatically. + + Process Flow: + 1. Retrieves or downloads the transformation model using caching mechanism + 2. Applies the transformation model to the input data + 3. Converts the transformed data to dictionary format for API response + 4. Handles any exceptions and returns structured error information + 5. Manages model lifecycle based on retention policy (cleanup artifacts if needed) + + Parameters: + model_name (str): The name of the MLFlow model to use for transformation. + data (pd.DataFrame): The input data to be transformed by the model. + model_retention (int): Cache retention time in minutes (0 = no caching). + + Returns: + dict: Response dictionary containing: + - success (bool): Operation success status + - content (dict): Transformed data as dictionary, or error information + if operation failed. Error content includes: + - message (str): Error description + - traceback (str): Full exception traceback + + Raises: + Exception: Any exception during model loading or transformation is caught + and returned in the response structure rather than propagated. + """ + + try: + + return { + 'success': True, + 'content': self.get_cached_transform( + model_name, data, model_retention).to_dict() + } + + except Exception as e: + return { + 'success': False, + 'content': { + 'message': str(e), + 'traceback': traceback.format_exc() + } + } + + def predict(self, model_name: str, data: pd.DataFrame, model_retention: int): + """ + Generate predictions using a cached prediction model. + + This method provides a high-level interface for model prediction operations + using MLFlow models. It handles model caching, performance monitoring, + response formatting, and error management automatically. + + Process Flow: + 1. Preserves input data index for result alignment + 2. Records prediction start time for performance measurement + 3. Retrieves or downloads the prediction model using caching mechanism + 4. Executes model prediction on the input data + 5. Formats predictions into DataFrame with proper column naming + 6. Restores original data index to maintain data alignment + 7. Calculates and adds response time measurement + 8. Converts results to dictionary format for API response + 9. Handles any exceptions and returns structured error information + + Parameters: + model_name (str): The name of the MLFlow model to use for prediction. + data (pd.DataFrame): The input data to make predictions on. + model_retention (int): Cache retention time in minutes (0 = no caching). + + Returns: + dict: Response dictionary containing: + - success (bool): Operation success status + - content (dict): Prediction results as dictionary with: + - prediction: Model predictions array + - response_time: Prediction execution time in seconds + Or error information if operation failed: + - message (str): Error description + - traceback (str): Full exception traceback + + Raises: + Exception: Any exception during model loading or prediction is caught + and returned in the response structure rather than propagated. + """ + try: + + input_index = data.index + start_time = datetime.now() + data = self.get_cached_predict( + model_name, data, model_retention) + + end_time = datetime.now() + data = pd.DataFrame(data, columns=['prediction']) + data.index = input_index + data['response_time'] = (end_time - start_time).total_seconds() + + return { + 'success': True, + 'content': data.to_dict() + } + + except Exception as e: + return { + 'success': False, + 'content': { + 'message': str(e), + 'traceback': traceback.format_exc() + } + } + + def retrain_model(self, data: pd.DataFrame, model_name: str) -> tuple: + """ + Orchestrate the complete model retraining workflow. + + This method coordinates the entire model retraining process by managing + the MLFlow experiment lifecycle, model loading, retraining execution, + and artifact management. It provides a comprehensive retraining solution + that maintains model versioning and experiment tracking. + + Process Flow: + 1. Creates MLFlow experiment environment: + - Loads current production prediction model + - Loads current production transformation model + - Fits transformation model with new training data + - Prepares transformed data for prediction model retraining + - Sets up MLFlow experiment context + 2. Executes model retraining: + - Starts new MLFlow run with descriptive metadata + - Logs model parameters and hyperparameters + - Retrains both prediction and transformation models + - Logs training data as artifacts + - Saves retrained models to MLFlow registry + - Cleans up temporary files + 3. Returns comprehensive retraining results + + Args: + data (pd.DataFrame): Training data for model retraining. Must contain + all features required by both transformation and + prediction models, including target variable. + model_name (str): Name of the MLFlow model to retrain. Must exist + in the MLFlow Model Registry in Production stage. + + Returns: + tuple: Retraining operation results containing: + - status_message (str): Success confirmation message or error details + - experiment_name (str): MLFlow experiment identifier for tracking + + Raises: + mlflow.exceptions.MlflowException: If model not found in registry + ValueError: If experiment cannot be created or models cannot be loaded + Exception: Any other exception during the retraining process + """ + prediction_model, data_model, experiment = self.create_model_experiment( + model_name, data) + retrain_result = self.perform_model_retrain( + prediction_model, data_model, experiment, model_name, data) + return retrain_result + def update_production_model(self, experiment: str, model_name: str) -> dict: """ Update production model using the latest retraining run. - This method orchestrates the complete production model update - process by identifying the most recent retraining run and - promoting it to production stage. + This method orchestrates the complete production model update process by + identifying the most recent retraining run and promoting it to production + stage. It handles model registration, versioning, and stage transitions + with comprehensive metadata tracking. + + Process Flow: + 1. Retrieves experiment information: + - Converts experiment name to MLFlow experiment ID + - Searches for the most recent retraining run in the experiment + - Filters runs by 'retrain' parameter and orders by completion time + 2. Promotes model to production: + - Registers the model from the specified run to MLFlow Model Registry + - Retrieves the latest model version number + - Transitions the model to 'Production' stage + - Archives existing production versions automatically + 3. Returns comprehensive update metadata Args: - experiment (str): MLFlow experiment name - model_name (str): Name of the MLFlow model + experiment (str): MLFlow experiment name containing the retraining runs. + Must be a valid experiment that exists in MLFlow. + model_name (str): Name of the MLFlow model to update. Must exist + in the MLFlow Model Registry. Returns: dict: Complete model update metadata containing: - model_name (str): Name of the updated model - - version (str): New model version number - - mlflow_run_id (str): Source run ID - - mlflow_experiment_id (int): Experiment ID + - version (str): New model version number (incremented automatically) + - mlflow_run_id (str): Source run ID of the promoted model + - mlflow_experiment_id (int): Experiment ID for tracking + + Raises: + ValueError: If experiment not found or model versions are invalid + mlflow.exceptions.MlflowException: If model registration or stage + transition fails + Exception: Any other exception during the update process + + Note: + This operation is irreversible. The previous production model will + be automatically archived when the new version is promoted. """ experiment_id = self.get_experiment(experiment) run_id = self.get_experiment_last_run(experiment_id) diff --git a/model_convert.ipynb b/model_convert.ipynb new file mode 100644 index 0000000..42f6c3d --- /dev/null +++ b/model_convert.ipynb @@ -0,0 +1,106 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 23, + "id": "e838ff21", + "metadata": {}, + "outputs": [], + "source": [ + "import csv\n", + "\n", + "def csv_to_tag_lists(csv_path: str) -> dict:\n", + " read_tags = []\n", + " write_tags = []\n", + "\n", + " def to_float(val):\n", + " try:\n", + " return float(str(val).strip())\n", + " except Exception:\n", + " return None\n", + "\n", + " with open(csv_path, newline=\"\", encoding=\"utf-8\") as f:\n", + " reader = csv.DictReader(f)\n", + " for row in reader:\n", + " # Basic normalization\n", + " op = (row.get(\"operation\") or \"\").strip()\n", + "\n", + " if op == \"READ\":\n", + " # Build common tag payload with required mappings\n", + " tag = {\n", + " \"server_id\": \"1\",\n", + " \"tag_address\": row.get(\"opc_tag\"),\n", + " \"tag_name\": row.get(\"name\"),\n", + " \"data_range\": [to_float(row.get(\"min_value\")), to_float(row.get(\"max_value\"))],\n", + " \"aggr_func\": row.get(\"aggregation_func\").lower(),\n", + " # keep other fields with their original names\n", + " \"frequency\": row.get(\"frequency\"),\n", + " \"local\": row.get(\"local\"),\n", + " \"area\": row.get(\"area\"),\n", + " \"description\": row.get(\"description\"),\n", + " }\n", + "\n", + " read_tags.append(tag)\n", + "\n", + " else:\n", + " tag = {\n", + " \"server_id\": \"1\",\n", + " \"addr\": row.get(\"opc_tag\"),\n", + " \"tag_name\": row.get(\"name\"),\n", + " \"local\": row.get(\"local\"),\n", + " \"area\": row.get(\"area\"),\n", + " \"description\": row.get(\"description\"),\n", + " }\n", + " \n", + " if op == \"WRITE_PREDICTION\":\n", + " tag[\"type\"] = \"prediction\"\n", + " write_tags.append(tag)\n", + " elif op == \"WRITE_CONFIDENCE\":\n", + " tag[\"type\"] = \"confidence\"\n", + " write_tags.append(tag)\n", + " # ignore any other operation values silently\n", + "\n", + " return {\"read_tags\": read_tags, \"write_tags\": write_tags}" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4621cd43", + "metadata": {}, + "outputs": [], + "source": [ + "import json\n", + "\n", + "file_names = [\"Courier - Página1.csv\"]\n", + "\n", + "for file_name in file_names:\n", + " write_file = file_name.replace(\".csv\", \".json\")\n", + "\n", + " with open(write_file, \"w\", encoding=\"utf-8\") as f:\n", + " json.dump(csv_to_tag_lists(file_name), f, indent=2, ensure_ascii=False)" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.11.13" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +}