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

@@ -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]:
         """

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,
+)

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

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,
             {

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,

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