SIENTIAPDE-1222

SIENTIAPDE-1214: Enhance MLFlow and tests with datetime index handling and logging improvements

- Added a new method in MLFlow to detect and parse datetime indices in DataFrames, ensuring proper format and raising errors for invalid types.
- Updated prediction workflows to utilize the new datetime index handling, improving data integrity during transformations.
- Enhanced logging in model_repository to include detailed data outputs for better traceability.
- Adjusted timeout settings in prediction workflows for improved execution time management.
- Updated tests.ipynb to include additional checks for index types and outputs for better validation of functionality.

laborious/activities/mlflow.py

@@ -1,8 +1,11 @@
-import json
 from temporalio import activity, workflow
 
 
 with workflow.unsafe.imports_passed_through():
+    from datetime import datetime
+    import json
+    from pandas import Timestamp, to_datetime
+    from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ
     from sientia_do.temporal.activities.base import BaseActivity
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
@@ -60,6 +63,44 @@ class MLFlow(BaseActivity):
             f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password, logger.base_logger
         )
 
+    def detect_and_parse_datetime_index(self, data: DataFrame, metadata: dict) -> DataFrame:
+        """
+        Detect and parse datetime index from data. index must be a timestamp like column.
+        This function must detect the timestamp type (pandas Timestamp or datetime) and convert it to DATETIME_FORMAT_WITH_TZ.
+        If the index is a string, must be in format DATETIME_FORMAT_WITH_TZ.
+        If another type or format, must raise an error.
+        """
+        index = data.index
+
+        # Get type of first element of index
+        index_type = type(index[0])
+
+        self.info(f"Index type: {index_type}", metadata)
+
+        message = f"Index must be all timestamp like column. Valid formats are: pandas Timestamp, datetime, string in format {DATETIME_FORMAT_WITH_TZ}"
+
+        # Check if all in index are of the same type
+        if not all(isinstance(i, index_type) for i in index):
+            raise ValueError(
+                f"{message}")
+
+        # Check type and converts to DATETIME_FORMAT_WITH_TZ
+        if index_type == str:
+            # Validate format of string and return error if not valid
+            try:
+                to_datetime(data.index)
+            except ValueError:
+                raise ValueError(
+                    f"{message}")
+
+        elif index_type == datetime or index_type == Timestamp:
+            data.index = data.index.strftime(DATETIME_FORMAT_WITH_TZ)
+        else:
+            raise ValueError(
+                f"{message}")
+
+        return data
+
     @activity.defn(name="request_transform")
     async def request_transform(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """
@@ -113,13 +154,43 @@ class MLFlow(BaseActivity):
         data.columns.name = None
 
         self.debug("Processed input data:", metadata)
-        self.debug(data, metadata)
+        data.to_csv('data.csv')
+        self.debug(data.to_string(), metadata)
 
         # Request transformation from MLFlow model
         response_data = self.model_monitoring_repository.transform(
             model_name, data, model_config
         )
 
+        self.debug("Raw response data:", metadata)
+        self.debug(response_data, metadata)
+
+        if response_data['success']:
+            response_dataframe = DataFrame(response_data['content'])
+            try:
+                response_dataframe = self.detect_and_parse_datetime_index(
+                    response_dataframe, metadata)
+                response_dataframe['timestamp'] = to_datetime(
+                    response_dataframe.index, format=DATETIME_FORMAT_WITH_TZ)
+                response_dataframe['timestamp'] = response_dataframe['timestamp'].dt.strftime(
+                    DATETIME_FORMAT)
+            except ValueError as e:
+                trace = traceback.format_exc()
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id='TRANSFORM_DATA_INDEX_ERROR',
+                    message=f'Error parsing trasnformed data index: {e}',
+                    block='transform',
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+                self.error(trace, metadata=metadata)
+                raise e
+
+            response_dataframe.to_csv('response_data.csv')
+
+            response_data['content'] = response_dataframe.to_dict()
+
         self.debug("Transform response data:", metadata)
         self.debug(response_data, metadata)