From 6ac0f38d59dad6da31a7911940f3a2d540926776 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 11 Nov 2025 16:50:22 -0300 Subject: [PATCH] SIENTIAPDE-1273 Update requirements and enhance metrics and data handling - Updated the sientia-dataops-library dependency version in requirements.txt to 1.5.3. - Added new metrics for model analysis, including lag, count, and error count in metrics.py. - Implemented a new method for formatting transformed data in gates.py. - Enhanced MLFlowRepository with methods to load artifact dataframes and calculate model metrics, including drift and performance metrics. - Updated the prediction process to handle transformed data and ensure proper execution of related activities in format_and_export_prediction.py and prediction_process.py. --- laborious/activities/gates.py | 22 +++++ laborious/metrics.py | 19 ++++ .../utils/repository/model_repository.py | 93 +++++++++++++++++++ .../format_and_export_prediction.py | 41 +++++++- .../sub_workflows/prediction_process.py | 2 + requirements.txt | 2 +- 6 files changed, 177 insertions(+), 2 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 9dc9eb1..56a1569 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -402,6 +402,28 @@ class Gates(SientiaMonitoring): return policy_type, int(policy_value) + + @activity.defn(name='format_transformed_data') + async def format_transformed_data(self, input_data: dict[str, Any]) -> dict[Any, Any]: + """ + Format transformed data according to configured storage policies. + """ + metadata = input_data['metadata'] + + model_id = input_data['model_id'] + + self.info('Formatting transformed data...', metadata) + + data = DataFrame(input_data['data']) + + data['timestamp'] = data.index + data = data.reset_index(drop=True) + + data = data.melt(id_vars='timestamp', var_name='variable', value_name='value') + data['model_id'] = model_id + + return data.to_dict() + @activity.defn(name='format_prediction') async def format_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]: """ diff --git a/laborious/metrics.py b/laborious/metrics.py index ec87867..59e4a81 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -169,3 +169,22 @@ MODEL_WRITE_ERROR_COUNT = Counter( 'Number of errors writing to the model', SIENTIA_CORE_LABELS, ) + +MODEL_ANALYZE_LAG = Histogram( + 'laborious_model_analyze_lag', + 'Lag between the start and end of analyze operations', + SIENTIA_CORE_LABELS, + buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0], +) + +MODEL_ANALYZE_COUNT = Counter( + 'laborious_model_analyze_count', + 'Number of analyze operations', + SIENTIA_CORE_LABELS, +) + +MODEL_ANALYZE_ERROR_COUNT = Counter( + 'laborious_model_analyze_error_count', + 'Number of errors during analyze operations', + SIENTIA_CORE_LABELS, +) \ No newline at end of file diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 1414713..edd9671 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -16,6 +16,7 @@ Capabilities: import ctypes import gc +from io import StringIO import threading import time import traceback @@ -33,6 +34,7 @@ from sientia_do.observability.logger import Logger from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ +from sientia.ModelAnalysis import ModelAnalysis from laborious import metrics @@ -259,6 +261,40 @@ class MLFlowRepository(SientiaMonitoring): return artifacts + + async def load_artifact_dataframe(self, model_name: str, artifact_path: str, + metadata: dict[str, Any]) -> pd.DataFrame: + + """ + Load the dataframe content of an artifact from the MLflow Model Registry. + + Args: + model_name (str): The name of the model to download from the registry. + artifact_path (str): The path to the artifact to load. + metadata (dict[str, Any]): Metadata used for structured logging. + + Returns: + pd.DataFrame: The dataframe content of the artifact. + """ + run_id = self.get_model_run_id(model_name=model_name, stage='Production') + artifact_path = path.join("runs:/", run_id, artifact_path) + core_labels = self.get_core_labels(metadata, operation_type='load_text') + + start_time = time.time() + try: + content = mlflow.artifacts.load_text(artifact_path) + except Exception as e: + await self.emit_metric(metric_object=metrics.MODEL_READ_ERROR_COUNT, tags=core_labels) + raise e + + await self.observe_lag(start_time, metrics.MODEL_READ_LAG, core_labels) + await self.emit_metric(metric_object=metrics.MODEL_READ_COUNT, tags=core_labels) + + dataframe = pd.read_csv(StringIO(content)) + + self.info(f'Loaded dataframe from {model_name}:{artifact_path}', metadata) + return dataframe + async def load_predict_model( self, model_name: str, metadata: dict[str, Any], flavor: str = 'sklearn' ) -> Any: @@ -1335,3 +1371,60 @@ class MLFlowRepository(SientiaMonitoring): metadata_result['mlflow_experiment_id'] = experiment_id return metadata_result + + async def get_model_metrics(self, + analyse_data: pd.DataFrame, + train_reference_data: pd.DataFrame, + test_reference_data: pd.DataFrame, + target_name: str, + drift_metrics: list[str], + performance_metrics: list[str], + metadata: dict) -> dict[str, Any]: + + """ + Calculates drift and preformance metrics for a model. + + Args: + analyse_data (pd.DataFrame): The data to analyse. + train_reference_data (pd.DataFrame): The train reference data. + test_reference_data (pd.DataFrame): The test reference data. + target_name (str): The target name. + drift_metrics (list[str]): The drift metrics. + performance_metrics (list[str]): The performance metrics. + metadata (dict): The metadata. + + Returns: + dict: The metrics. + """ + columns = train_reference_data.drop( + columns=[target_name, 'timestamp', 'target', 'prediction'], errors='ignore').columns + + config = { + "target": target_name, + "prediction": "prediction", + "timestamp": "timestamp", + "columns": columns, + } + + model_analysis = ModelAnalysis(config=config) + + core_labels = self.get_core_labels(metadata, operation_type='detect_univariate_drift') + + start_time = time.time() + try: + model_analysis.detect_univariate_drift( + reference_df=train_reference_data, + analyse_df=analyse_data, + features=columns, + timestamp_column=config['timestamp'], + metrics=drift_metrics, + chunk_period="s" + ) + except Exception as e: + await self.emit_metric(metric_object=metrics.MODEL_ANALYZE_ERROR_COUNT, tags=core_labels) + raise e + + await self.observe_lag(start_time, metrics.MODEL_ANALYZE_LAG, core_labels) + await self.emit_metric(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels) + + return {} diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 8d8c9da..bd28966 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -69,6 +69,7 @@ class FormatAndExportPrediction: metadata = input_data['metadata'] path_flag = input_data['path_flag'] data = input_data['data'] + transformed_data = input_data.get('transformed_data', None) prediction_confidence = input_data['prediction_confidence'] if path_flag is None: @@ -87,6 +88,37 @@ class FormatAndExportPrediction: start_to_close_timeout=timedelta(seconds=60), ) + if transformed_data is not None: + + transformed = await workflow.execute_local_activity_method( + Activities.format_transformed_data, + { + **metadata, + 'data': transformed_data, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60), + ) + + write_transformed_handler = workflow.execute_activity_method( + Activities.export_data_to_postgres, + { + **metadata, + 'schema': input_data['schema'], + 'table_name': input_data['transform_table_name'], + 'data': transformed, + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT_WITH_TZ, + }, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60), + ) + + else: + write_transformed_handler = None + else: # create default prediction prediction = await workflow.execute_local_activity_method( @@ -102,6 +134,8 @@ class FormatAndExportPrediction: start_to_close_timeout=timedelta(seconds=60), ) + write_transformed_handler = None + # write to opc prediction, opc_metrics = await workflow.execute_activity_method( Activities.write_opc_data, @@ -115,7 +149,7 @@ class FormatAndExportPrediction: ) # write to postgres - await workflow.execute_activity_method( + prediction_handler = workflow.execute_activity_method( Activities.export_data_to_postgres, { **metadata, @@ -128,6 +162,11 @@ class FormatAndExportPrediction: start_to_close_timeout=timedelta(seconds=180), ) + await prediction_handler + + if write_transformed_handler is not None: + await write_transformed_handler + await workflow.execute_activity_method( Activities.write_metrics, { diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index ab78fb1..fa07504 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -82,6 +82,7 @@ class PredictionProcess: model_id = input_data['model_id'] model_name = input_data['model_name'] model_config = input_data.get('model_config', {}) + save_transform = input_data.get('save_transform', True) # Get last timestamp for incremental processing last_timestamp = await workflow.execute_local_activity_method( @@ -199,6 +200,7 @@ class PredictionProcess: 'metadata': metadata, 'path_flag': path_flag, 'data': response_data['content'], + 'transformed_data': transformed_data if save_transform else None, 'prediction_confidence': confidence, 'timestamp': last_timestamp, 'model_id': model_id, diff --git a/requirements.txt b/requirements.txt index 12feb2e..750ef66 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,7 +3,7 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 prometheus-client botocore