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

SIENTIAPDE-1081

Browse Source 
Enhance documentation across multiple modules with detailed parameter descriptions and usage examples
This commit is contained in:
vitor-aignosi
2025-05-26 16:45:50 -03:00
parent bc3f26c280
commit 49fa60c66f
15 changed files with 358 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,9 +1,11 @@
"""
Model Monitoring Repository
This module contains the ModelMonitoringRepository class, which is responsible for handling the communication with the Model Monitoring API.
This module contains the ModelMonitoringRepository class,
which is responsible for handling the communication with the Model Monitoring API.
It includes the methods that are used to answer ModelMonitoringService requests using the Model Monitoring API functions.
It includes the methods that are used to answer ModelMonitoringService
requests using the Model Monitoring API functions.
By Monitoring we mean the evaluation of the performance of models, the generation of reports.
@@ -13,7 +15,6 @@ import traceback
import mlflow
import pandas as pd
from sientia.ModelServing import ModelServing
from pathlib import Path
class MLFlowRepository():
@@ -49,7 +50,8 @@ class MLFlowRepository():
def get_artifact(self, destination: str, search_by: str, run_id: str = None,
model_name: str = None, artifact_name: str = None) -> None:
"""
Get an artifact in MLflow by experiment or model and save it to a destination path using API.
Get an artifact in MLflow by experiment or model and save it to a
destination path using API.
If the artifact is searched by model, the latest production version will be used.
Args:
@@ -77,7 +79,9 @@ class MLFlowRepository():
Returns:
dict: the metrics of the model including MSE and R2
"""
return self.model_serving.get_model_metrics(reference_data=None, real_data=real_data, predictions=predictions, type_flag=flag)
return self.model_serving.get_model_metrics(
reference_data=None, real_data=real_data,
predictions=predictions, type_flag=flag)
def get_experiment_by_run_id(self, run_id: str) -> dict:
# Get the run information using the run_id
@@ -112,25 +116,26 @@ class MLFlowRepository():
Retrain a model with new data.
Parameters:
data (pandas.DataFrame): The new data to use for retraining.
model_name (str): The name of the model to retrain.
metrics_list (list): The metrics to be used to compare the models.
compare_metrics (bool): If True, the retrain will only be considered if the new model is better than the current one.
If False, the retrain will always be considered.
split_dataset (bool): If True, the data will be split into X and Y and into training and testing sets.
- data (pandas.DataFrame): The new data to use for retraining.
- model_name (str): The name of the model to retrain.
- metrics_list (list): The metrics to be used to compare the models.
- compare_metrics (bool): If True, the retrain will only be considered if
the new model is better than the current one.
If False, the retrain will always be considered.
- split_dataset (bool): If True, the data will be split into X and Y and into training and testing sets.
If False, the data will be used as a unique block for retraining.
update_report (bool): If True, a report will be created with the data of the retrained model.
update_transformation (bool): If True, the model will be updated in the MLflow tracking server.
update_prediction (bool): If True, the prediction model will be updated in the MLflow tracking server.
shuffle_data (bool): If True, the data will be shuffled before splitting.
model_type (str): The type of model to get metrics for. Ex: 'regression', 'classification'.
- update_report (bool): If True, a report will be created with the data of the retrained model.
- update_transformation (bool): If True, the model will be updated in the MLflow tracking server.
- update_prediction (bool): If True, the prediction model will be updated in the MLflow tracking server.
- shuffle_data (bool): If True, the data will be shuffled before splitting.
- model_type (str): The type of model to get metrics for. Ex: 'regression', 'classification'.
Returns:
mlflow.sklearn.Model: The retrained prediction model.
mlflow.sklearn.Model: The retrained data model.
mse (float): The mean squared error of the retrained model.
r2 (float): The R-squared score of the retrained model.
- mlflow.sklearn.Model: The retrained prediction model.
- mlflow.sklearn.Model: The retrained data model.
- mse (float): The mean squared error of the retrained model.
- r2 (float): The R-squared score of the retrained model.
"""
# load predictor model
@@ -218,13 +223,16 @@ class MLFlowRepository():
def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict:
# Registrar o modelo
# Aqui estamos assumindo que você já tem um modelo salvo, caso contrário você precisará treiná-lo e salvá-lo primeiro.
# Se o modelo já está registrado, você pode usar o método register_model() ou pyfunc.load_model() para isso.
# Aqui estamos assumindo que você já tem um modelo salvo,
# caso contrário você precisará treiná-lo e salvá-lo primeiro.
# Se o modelo já está registrado, você pode usar o método
# register_model() ou pyfunc.load_model() para isso.
mlflow.register_model(
f"runs:/{run_id}/prediction_model", model_name)
# Colocar a versão do modelo em produção
# Depois de registrar o modelo, precisamos pegar a versão mais recente do modelo e movê-lo para o estágio 'Production'
# Depois de registrar o modelo, precisamos pegar a versão mais
# recente do modelo e movê-lo para o estágio 'Production'
client = mlflow.tracking.MlflowClient()
# Obter a versão mais recente registrada do modelo
@@ -257,10 +265,23 @@ class MLFlowRepository():
return metadata
def transform(self, model_name: str, data: pd.DataFrame, model_retention: int):
"""
Transform data using a model.
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.
Returns:
- dict: A dictionary containing the transformed data.
"""
try:
return {
'success': True,
'content': self.model_serving.get_cached_transform(model_name, data, model_retention).to_dict()
'content': self.model_serving.get_cached_transform(
model_name, data, model_retention).to_dict()
}
except Exception as e:
@@ -273,6 +294,17 @@ class MLFlowRepository():
}
def predict(self, model_name: str, data: pd.DataFrame, model_retention: int):
"""
Predict data using a model.
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.
Returns:
- dict: A dictionary containing the predicted data.
"""
try:
start_time = datetime.now()
data = self.model_serving.get_cached_predict(

53
laborious/utils/repository/opc_repository.py
View File

@@ -1,12 +1,12 @@
import traceback
from logging import Logger
from datetime import datetime
from pathlib import Path
from asyncua.sync import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256
from asyncua.ua import DataValue, Variant, VariantType
from logging import Logger
from datetime import datetime
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.notifications.models import NotificationLevel
import traceback
data_type_map = {
'float': {
@@ -33,7 +33,8 @@ data_type_map = {
class OpcRepository():
def __init__(self, name: str, url: str, logger: Logger, notification_handler: NotificationHandler,
def __init__(self, name: str, url: str, logger: Logger,
notification_handler: NotificationHandler,
reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None,
private_key_path: str = None, server_cert_path: str = None):
self.url = url
@@ -57,12 +58,12 @@ class OpcRepository():
Raises:
ValueError: If either the certificate path or private key path is not provided.
Attributes:
cert_path (str): Path to the client's certificate file.
private_key_path (str): Path to the client's private key file.
server_cert_path (str, optional): Path to the server's certificate file.
server_uri (str): The URI of the server to be used as the application URI.
client (opcua.Client): The OPC UA client instance.
logger (logging.Logger): Logger instance for logging information.
- cert_path (str): Path to the client's certificate file.
- private_key_path (str): Path to the client's private key file.
- server_cert_path (str, optional): Path to the server's certificate file.
- server_uri (str): The URI of the server to be used as the application URI.
- client (opcua.Client): The OPC UA client instance.
- logger (logging.Logger): Logger instance for logging information.
Security Settings:
- Security Policy: Basic256
- Secure Channel Timeout: 10,000,000 ms
@@ -105,6 +106,12 @@ class OpcRepository():
return self.try_connect()
def try_connect(self):
"""
Tries to connect to the OPC server.
Returns:
bool: True if the connection was successful, False otherwise.
"""
try:
self.last_reconnection_time = datetime.now()
self.client.connect()
@@ -122,6 +129,9 @@ class OpcRepository():
return False
def disconnect(self):
"""
Disconnects from the OPC server.
"""
if self.client is None:
return
self.client.disconnect()
@@ -129,12 +139,25 @@ class OpcRepository():
self.logger.info('Disconnected from OPC server')
def __del__(self):
"""
Disconnects from the OPC server when the object is destroyed.
"""
try:
self.disconnect()
except Exception as e:
self.logger.error(f"Error in destructor: {e}")
def validate_connection(self):
"""
Validates the connection to the OPC server.
If the connection is not established, it attempts to reconnect.
If the connection is established but the client is not connected,
it attempts to reconnect.
If the connection is established but the client is connected,
it checks if the client is connected to the OPC server.
If the client is not connected, it attempts to reconnect.
If the client is connected, it returns True.
"""
if self.client is None:
return self.connect()
@@ -168,6 +191,16 @@ class OpcRepository():
return True
def write_data(self, node, value, data_type):
"""
Writes data to the OPC server.
If the connection is not established, it attempts to reconnect.
If the connection is established but the client is not connected,
it attempts to reconnect.
If the connection is established but the client is connected,
it checks if the client is connected to the OPC server.
If the client is not connected, it attempts to reconnect.
If the client is connected, it returns True.
"""
if not self.validate_connection():
return
try: