SIENTIAPDE-1712

SIENTIAPDE-1712 Refactor debug logging for DataFrames across multiple classes. Introduced a new method to log DataFrame content conditionally based on row count in Gates, MLFlow, ModelMetrics, and MLFlowRepository classes, improving debugging capabilities while managing log output effectively.

laborious/activities/gates.py

@@ -16,6 +16,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.utils.formatters import create_sample_dict
 
     from laborious import metrics
+    from laborious.utils.dataframe_debug import build_dataframe_debug_message
     from laborious.utils.filters.conditional_filters import (
         filter_empty_data,
         filter_specific_variables_null_values,
@@ -86,6 +87,7 @@ class Gates(MinioManager):
     """
 
     minio_repository: MinioRepository | None = None
+    _MAX_DEBUG_DATAFRAME_ROWS = 100
 
     def __init__(
         self,
@@ -118,6 +120,24 @@ class Gates(MinioManager):
     def __del__(self):
         self.close()
 
+    def _debug_dataframe(self, message: str, data: Any, metadata: dict[str, Any]) -> None:
+        """
+        Log dataframe content only when row count is below the configured threshold
+
+        Args:
+         - message (str): Base log message to identify the dataframe in logs
+         - data (Any): Dataframe-like payload to be logged
+         - metadata (dict[str, Any]): Workflow metadata for contextual logging
+        """
+        self.debug(
+            build_dataframe_debug_message(
+                message=message,
+                data=data,
+                max_rows=self._MAX_DEBUG_DATAFRAME_ROWS,
+            ),
+            metadata,
+        )
+
     @staticmethod
     def _read_filter_entry(config: dict[str, Any]) -> tuple[str, dict[str, Any]]:
         """
@@ -179,7 +199,7 @@ class Gates(MinioManager):
 
         filter_output = []
 
-        self.debug(f'Input data: {data.head(5).to_string()}', metadata)
+        self._debug_dataframe('Input data:', data, metadata)
         self.debug(f'Filters: {filters}', metadata)
 
         # Apply each configured filter
@@ -355,7 +375,7 @@ class Gates(MinioManager):
 
         filter_output = []
 
-        self.debug(f'Input data:\n {data.head(5).to_string()}', metadata)
+        self._debug_dataframe('Input data:', data, metadata)
         self.debug(f'Filters: \n {filters}', metadata)
 
         for fil, config in filters.items():
@@ -544,7 +564,7 @@ class Gates(MinioManager):
         data = data.reset_index(drop=True)
 
         self.debug(f'Prediction store policy: {prediction_store_policy}', metadata)
-        self.debug(f'Prediction data: {data.head(5).to_string()}', metadata)
+        self._debug_dataframe('Prediction data:', data, metadata)
 
         policy_type, policy_value = self.get_prediction_store_policy(
             prediction_store_policy, metadata
@@ -581,7 +601,7 @@ class Gates(MinioManager):
         data = data.reset_index(drop=True)
 
         self.info(f'Prediction formatted: {len(data)} rows', metadata)
-        self.debug(f'Prediction data: {data.head(5).to_string()}', metadata)
+        self._debug_dataframe('Prediction data:', data, metadata)
 
         return data.to_dict()
 
@@ -696,7 +716,7 @@ class Gates(MinioManager):
             report['mlflow_run_id'] = update_report['mlflow_run_id']
             report['mlflow_experiment_id'] = update_report['mlflow_experiment_id']
 
-        self.debug(f'Retrain report: {report.to_csv()}', metadata)
+        self._debug_dataframe('Retrain report:', report, metadata)
 
         return report.to_dict()
 

laborious/activities/mlflow.py

@@ -20,6 +20,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.utils.formatters import create_sample_dict
 
     from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload
+    from laborious.utils.dataframe_debug import build_dataframe_debug_message
     from laborious.utils.repository.minio_manager import MinioManager
     from laborious.utils.repository.model_repository import MLFlowRepository
 
@@ -104,19 +105,11 @@ class MLFlow(MinioManager):
          - data (Any): Dataframe-like object expected to expose shape and to_csv
          - metadata (dict[str, Any]): Workflow metadata for contextual logging
         """
-        if not hasattr(data, 'shape') or not hasattr(data, 'to_csv'):
-            self.debug(f'{message}\n{data}', metadata)
-            return
-
-        rows = data.shape[0]
-        if rows <= self._MAX_DEBUG_DATAFRAME_ROWS:
-            self.debug(f'{message}\n{data.to_csv()}', metadata)
-            return
-
         self.debug(
-            (
-                f'{message} skipped because dataframe has {rows} rows '
-                f'(max: {self._MAX_DEBUG_DATAFRAME_ROWS}). Shape: {data.shape}'
+            build_dataframe_debug_message(
+                message=message,
+                data=data,
+                max_rows=self._MAX_DEBUG_DATAFRAME_ROWS,
             ),
             metadata,
         )

laborious/activities/model_metrics.py

@@ -17,6 +17,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ
 
     from laborious import metrics
+    from laborious.utils.dataframe_debug import build_dataframe_debug_message
 
 warnings.filterwarnings('ignore', category=RuntimeWarning, message='Degrees of freedom <= 0')
 warnings.filterwarnings(
@@ -26,6 +27,8 @@ warnings.filterwarnings(
 
 class ModelMetrics(SientiaMonitoring):
     """
+    _MAX_DEBUG_DATAFRAME_ROWS = 100
+
     Metrics activities for the Laborious system.
 
     This class provides activities for writing metrics to the Prometheus monitoring system.
@@ -48,6 +51,24 @@ class ModelMetrics(SientiaMonitoring):
     def __del__(self):
         self.close()
 
+    def _debug_dataframe(self, message: str, data: Any, metadata: dict[str, Any]) -> None:
+        """
+        Log dataframe content only when row count is below the configured threshold
+
+        Args:
+         - message (str): Base log message to identify the dataframe in logs
+         - data (Any): Dataframe-like payload to be logged
+         - metadata (dict[str, Any]): Workflow metadata for contextual logging
+        """
+        self.debug(
+            build_dataframe_debug_message(
+                message=message,
+                data=data,
+                max_rows=self._MAX_DEBUG_DATAFRAME_ROWS,
+            ),
+            metadata,
+        )
+
     async def get_drift_metrics(
         self,
         reference_data: DataFrame,
@@ -78,14 +99,9 @@ class ModelMetrics(SientiaMonitoring):
 
         model_analysis = ModelAnalysis(config=config)
 
-        self.debug(
-            f'Reference data: Size {reference_data.shape} \n{reference_data.head(5).to_string()}',
-            metadata,
-        )
+        self._debug_dataframe(f'Reference data: Size {reference_data.shape}', reference_data, metadata)
 
-        self.debug(
-            f'Target data: Size {target_data.shape} \n{target_data.head(5).to_string()}', metadata
-        )
+        self._debug_dataframe(f'Target data: Size {target_data.shape}', target_data, metadata)
 
         core_labels = self.get_core_labels(metadata, operation_type='detect_univariate_drift')
         start_time = time.time()
@@ -142,9 +158,7 @@ class ModelMetrics(SientiaMonitoring):
         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)
 
-        self.debug(
-            f'Drift dataframe: Size {drift_df.shape} \n{drift_df.head(5).to_string()}', metadata
-        )
+        self._debug_dataframe(f'Drift dataframe: Size {drift_df.shape}', drift_df, metadata)
 
         return drift_df
 
@@ -274,11 +288,7 @@ class ModelMetrics(SientiaMonitoring):
         drift_df['timestamp'] = drift_df['timestamp'].dt.tz_localize('UTC')
         drift_df['timestamp'] = drift_df['timestamp'].dt.strftime(DATETIME_FORMAT_WITH_TZ)
 
-        self.debug(
-            f'Drift dataframe: Size {drift_df.shape} \n{drift_df.head(5).to_string()}', metadata
-        )
-
-        self.debug(f'Drift dataframe: {drift_df.head(5).to_string()}', metadata)
+        self._debug_dataframe(f'Drift dataframe: Size {drift_df.shape}', drift_df, metadata)
 
         return drift_df.to_dict(orient='records')
 
@@ -347,8 +357,6 @@ class ModelMetrics(SientiaMonitoring):
         data['data_size'] = data_size
         data['interval_minutes'] = interval_minutes
 
-        self.debug(
-            f'Simple metrics dataframe: Size {data.shape} \n{data.head(5).to_string()}', metadata
-        )
+        self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata)
 
         return data.to_dict(orient='records')