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

SIENTIAPDE-1231

Browse Source 
Refactor OPC and model repository for improved functionality and clarity

- Updated OPC server logging to handle missing prediction and confidence tags gracefully.
- Corrected documentation for OPC reconnection interval from milliseconds to seconds.
- Enhanced MLFlowRepository with new methods for model retrieval, caching, and transformation, improving model management and retraining workflows.
This commit is contained in:
vitor-aignosi
2025-09-23 16:06:10 -03:00
parent 8340189741
commit ec0a193c32
6 changed files with 969 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
encode.sh Executable file
View File

@@ -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 .

112
encrypt.py Normal file
View File

@@ -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()

2
laborious/activities/opc.py
View File

@@ -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)

2
laborious/utils/connectors_config.py
View File

@@ -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

899
laborious/utils/repository/model_repository.py
View File

@@ -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)
environ["MLFLOW_TRACKING_USERNAME"] = username
environ["MLFLOW_TRACKING_PASSWORD"] = password
# Create an MLflow client
self.client = mlflow.tracking.MlflowClient()
self.model_cache = {}
def transform(self, model_name: str, data: pd.DataFrame, model_retention: int):
"""
Transform data using a model.
Functions related to get model registry parameters
"""
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.
def get_model_uri(self, run_id: str, prediction: bool = True):
"""
Get the model URI based on the run_id.
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)

106
model_convert.ipynb Normal file
View File

@@ -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
}