diff --git a/README.md b/README.md
index 0143cbd..a34b939 100644
--- a/README.md
+++ b/README.md
@@ -130,8 +130,14 @@ Laborious uses a Temporal-based architecture with strong separation of concerns
- `minimal_retrain.py`: Automated model retraining and production update
#### **Activities (`laborious/activities/`)**
-- `gates.py`: Data quality validation and filtering
-- `mlflow.py`: Transform and predict operations
+- `gates.py`: Data quality validation, filtering, and data formatting operations
+ - Input/response/content gates for quality validation
+ - Prediction and transformed data formatting
+ - Retrain report formatting and metrics recording
+- `mlflow.py`: Transform, predict, and model management operations
+ - MLFlow model transformation and prediction
+ - Model retraining and production updates
+ - Reference data retrieval from MLflow Model Registry
- `opc.py`: OPC UA export to industrial systems (optional)
- `activities.py`: Aggregates activity interfaces
@@ -146,7 +152,9 @@ Laborious uses a Temporal-based architecture with strong separation of concerns
#### **1. Batch Prediction Pipeline**
```
Input Data (PostgreSQL) → Data Quality Gates → MLFlow Transform →
-MLFlow Prediction → Response Validation → Export (PostgreSQL [+ OPC])
+MLFlow Prediction → Response Validation → Format & Export
+ ├─→ Predictions → PostgreSQL [+ OPC]
+ └─→ Transformed Data → PostgreSQL (optional)
```
#### **2. Model Retraining Pipeline**
@@ -318,27 +326,37 @@ The **FormatAndExportPrediction** workflow handles prediction data formatting an
#### Execution Flow
1. **Path Decision**: Determines formatting path based on configuration
2. **Data Formatting**: Formats prediction data for specific output requirements
-3. **PostgreSQL Export**: Writes formatted predictions to database
-4. **OPC Export**: Writes predictions to OPC servers
-5. **Metrics Recording**: Records export performance and success metrics
+3. **Transformed Data Processing**: Optionally formats and exports transformed data separately
+4. **PostgreSQL Export**: Writes formatted predictions to database
+5. **OPC Export**: Writes predictions to OPC servers
+6. **Metrics Recording**: Records export performance and success metrics
#### Key Features
- **Flexible Formatting**: Configurable output formats for different destinations
- **Multi-Destination Export**: PostgreSQL and OPC server integration
+- **Transformed Data Export**: Optional separate export of MLFlow transformed data
- **Performance Monitoring**: Comprehensive metrics for export operations
- **Error Handling**: Robust error handling with notification integration
#### Architecture Diagram
```mermaid
flowchart LR
- A[1. format_prediction/format_default_prediction] --> B[2. write_opc_data] --> C[3. export_data_to_postgres] --> D[4. write_metrics]
+ A[1. format_prediction/format_default_prediction] --> B[2. format_transformed_data] --> C[3. write_opc_data] --> D[4. export_data_to_postgres] --> E[5. write_metrics]
A -.-> Format[Data Formatting]
- B -.-> OPC[OPC Servers]
- C -.-> PostgreSQL[(PostgreSQL)]
- D -.-> Prometheus[Prometheus]
+ B -.-> Transform[Transformed Data]
+ C -.-> OPC[OPC Servers]
+ D -.-> PostgreSQL[(PostgreSQL)]
+ E -.-> Prometheus[Prometheus]
```
+#### Transformed Data Export
+When `transformed_data` is provided in the input, the workflow will:
+- Format the transformed data using `format_transformed_data` activity
+- Export it to a separate table (`transform_table_name`) asynchronously
+- Wait for both prediction and transformed data exports to complete
+- This enables separate tracking of model transformations for analysis and debugging
+
### 4. Minimal Retrain Workflow (`minimal_retrain.py`)
The **MinimalRetrain** workflow handles automated model retraining and production model updates.
@@ -579,11 +597,24 @@ The workflow at `.github/workflows/quality-gate.yml` executes validations on eac
```
tests/
├── activities/ # Activity implementation tests
-├── workflow/ # Workflow orchestration tests
+│ ├── test_gates.py # Data quality gates and formatting tests
+│ ├── test_mlflow.py # MLFlow operations and reference data tests
+│ └── ... # Other activity tests
+├── workflows/ # Workflow orchestration tests
+│ └── subworkflows/ # Sub-workflow tests
+│ └── test_format_and_export_prediction.py # Export workflow tests
├── utils/ # Utility function tests
└── integration/ # End-to-end workflow tests
```
+### Test Coverage
+The test suite provides comprehensive coverage for:
+- **Data Quality Gates**: Input, response, and content validation filters
+- **Data Formatting**: Prediction, transformed data, and retrain report formatting
+- **MLFlow Operations**: Transform, predict, retrain, and reference data retrieval
+- **Workflow Orchestration**: Complete workflow execution paths and error handling
+- **Metrics Recording**: Performance monitoring and OPC export metrics
+
### Test Execution
```bash
# Install test dependencies
diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py
index e09d95f..077a433 100644
--- a/laborious/activities/activities.py
+++ b/laborious/activities/activities.py
@@ -9,11 +9,12 @@ with workflow.unsafe.imports_passed_through():
from laborious.activities.gates import Gates
from laborious.activities.mlflow import MLFlow
+ from laborious.activities.model_metrics import ModelMetrics
from laborious.activities.opc import OPC
from laborious.activities.storage import Storage
-class Activities(Storage, MLFlow, Gates, OPC):
+class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics):
"""
Main activities orchestrator for the Laborious system.
@@ -108,6 +109,13 @@ class Activities(Storage, MLFlow, Gates, OPC):
metrics_controller=metrics_controller,
)
+ ModelMetrics.__init__(
+ self,
+ logger=logger,
+ notification_handler=notification_handler,
+ metrics_controller=metrics_controller,
+ )
+
async def shutdown(self):
"""
Gracefully shutdown all activities and clean up resources.
@@ -124,3 +132,4 @@ class Activities(Storage, MLFlow, Gates, OPC):
MLFlow.close(self)
Gates.close(self)
await OPC.close(self)
+ ModelMetrics.close(self)
diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py
index 9dc9eb1..f0f8474 100644
--- a/laborious/activities/gates.py
+++ b/laborious/activities/gates.py
@@ -402,8 +402,54 @@ 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:
+ """
+ Format transformed data for storage and export operations.
+
+ This method formats transformed data from MLFlow model transformations
+ into a standardized format suitable for database storage. It converts
+ wide-format data (columns as variables) into long-format (melted)
+ with proper timestamp handling and model identification.
+
+ The formatting process includes:
+ 1. Converting input data dictionary to DataFrame
+ 2. Extracting timestamps from DataFrame index
+ 3. Resetting index to create sequential row numbers
+ 4. Melting data from wide format to long format (variable-value pairs)
+ 5. Adding model_id for data lineage tracking
+
+ Args:
+ input_data (dict): Input data containing:
+ - metadata (dict): Workflow execution metadata
+ - data (dict[str, Any]): Transformed data to format (DataFrame-compatible dict)
+ - model_id (str): Unique identifier for the ML model
+
+ Returns:
+ dict: Formatted data dictionary with keys:
+ - timestamp (dict): Timestamp values indexed by row number
+ - variable (dict): Variable names indexed by row number
+ - value (dict): Variable values indexed by row number
+ - model_id (dict): Model identifiers indexed by row number
+ """
+ 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]:
+ async def format_prediction(self, input_data: dict[str, Any]) -> dict:
"""
Format prediction data according to configured storage policies.
@@ -476,7 +522,7 @@ class Gates(SientiaMonitoring):
return data.to_dict()
@activity.defn(name='format_default_prediction')
- async def format_default_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]:
+ async def format_default_prediction(self, input_data: dict[str, Any]) -> dict:
"""
Create and format default prediction data for error conditions.
@@ -521,9 +567,47 @@ class Gates(SientiaMonitoring):
return data.to_dict()
@activity.defn(name='format_retrain_report')
- async def format_retrain_report(self, input_data: dict[str, Any]) -> dict[Any, Any]:
+ async def format_retrain_report(self, input_data: dict[str, Any]) -> dict:
"""
- Format retrain report data according to configured storage policies.
+ Format retrain report data for storage and audit trail maintenance.
+
+ This method formats model retraining operation results into a standardized
+ report format suitable for database storage and operational monitoring.
+ It captures retraining status, timestamps, and model version information
+ for comprehensive audit trails and operational visibility.
+
+ The formatting process includes:
+ 1. Extracting retraining experiment response data
+ 2. Capturing model update report information (version, MLflow IDs)
+ 3. Formatting timestamps and status information
+ 4. Conditionally including version information for successful retrains
+
+ Args:
+ input_data (dict): Input data containing:
+ - metadata (dict): Workflow execution metadata
+ - experiment_response (dict): Retraining experiment response containing:
+ - success (bool): Retraining operation success status
+ - timestamp (str): Timestamp of the retraining operation
+ - message (str): Status message or error description
+ - update_report (dict): Model update report containing:
+ - version (str): New model version identifier
+ - mlflow_run_id (str): MLflow run identifier
+ - mlflow_experiment_id (str): MLflow experiment identifier
+ - model_id (str): Unique identifier for the ML model
+ - model_name (str): Name of the ML model
+
+ Returns:
+ dict: Formatted retrain report dictionary with keys:
+ - model_id (dict): Model identifiers indexed by row number
+ - model_name (dict): Model names indexed by row number
+ - timestamp (dict): Retraining timestamps indexed by row number
+ - status (dict): Retraining status messages indexed by row number
+ - version (dict, optional): Model versions indexed by row number
+ Only included if experiment_response['success'] is True
+ - mlflow_run_id (dict, optional): MLflow run IDs indexed by row number
+ Only included if experiment_response['success'] is True
+ - mlflow_experiment_id (dict, optional): MLflow experiment IDs indexed by row number
+ Only included if experiment_response['success'] is True
"""
metadata = input_data['metadata']
self.info('Formatting retrain report...', metadata)
diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py
index 98be4f2..3970bb7 100644
--- a/laborious/activities/mlflow.py
+++ b/laborious/activities/mlflow.py
@@ -418,3 +418,51 @@ class MLFlow(SientiaMonitoring):
)
self.error(trace, metadata=metadata)
raise e
+
+ @activity.defn(name='get_reference_data')
+ async def get_reference_data(self, input_data: dict[str, Any]) -> list[dict] | None:
+ """
+ Get reference data from the MLflow Model Registry.
+
+ This method retrieves evaluation reference data stored as artifacts in the
+ MLflow Model Registry. The reference data is typically used for model
+ drift detection, performance comparison, and quality validation. The method
+ loads the data from a CSV artifact file and formats timestamps for
+ consistent processing.
+
+ The method handles:
+ 1. Loading evaluation data artifact from MLflow Model Registry
+ 2. Timestamp parsing and formatting for consistency
+ 3. Data conversion to dictionary format for workflow consumption
+ 4. Graceful handling of missing reference data
+
+ Args:
+ input_data (dict): Input data containing:
+ - metadata (dict): Workflow execution metadata
+ - model_name (str): Name of the MLFlow model to get reference data from
+
+ Returns:
+ list[dict[Hashable, Any]] | None: Reference data from the MLflow Model Registry
+ as a list of dictionaries. Returns None if reference data is not found
+ or if the artifact does not exist.
+
+ Raises:
+ Exception: If artifact loading fails or encounters errors during processing
+ """
+
+ metadata = input_data['metadata']
+ model_name = input_data['model_name']
+ artifact = 'evaluation_data.csv'
+
+ reference_data = await self.model_monitoring_repository.load_artifact_dataframe(
+ model_name=model_name, artifact_path=artifact, metadata=metadata
+ )
+
+ if reference_data is None:
+ self.warning(f'Reference data not found for model {model_name}', metadata)
+ return None
+
+ reference_data['timestamp'] = to_datetime(reference_data['timestamp'])
+ reference_data['timestamp'] = reference_data['timestamp'].dt.strftime(DATETIME_FORMAT)
+
+ return reference_data.to_dict(orient='records')
diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py
new file mode 100644
index 0000000..29b2848
--- /dev/null
+++ b/laborious/activities/model_metrics.py
@@ -0,0 +1,354 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+ import time
+ import traceback
+ import warnings
+ from typing import Any
+
+ import numpy as np
+ from pandas import DataFrame, Index, to_datetime
+ from sientia.ModelAnalysis import ModelAnalysis
+ from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+ from sientia_do.notifications.models import NotificationLevel
+ 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, DATETIME_FORMAT_WITH_TZ
+
+ from laborious import metrics
+
+warnings.filterwarnings('ignore', category=RuntimeWarning, message='Degrees of freedom <= 0')
+warnings.filterwarnings(
+ 'ignore', category=RuntimeWarning, message='invalid value encountered in scalar divide'
+)
+
+
+class ModelMetrics(SientiaMonitoring):
+ """
+ Metrics activities for the Laborious system.
+
+ This class provides activities for writing metrics to the Prometheus monitoring system.
+ """
+
+ def __init__(
+ self,
+ logger: Logger,
+ notification_handler: NotificationHandler,
+ metrics_controller: MetricsController,
+ ):
+ SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
+
+ def close(self) -> None:
+ """
+ Close the model metrics activity and clean up resources.
+ """
+ SientiaMonitoring.shutdown(self)
+
+ def __del__(self):
+ self.close()
+
+ async def get_drift_metrics(
+ self,
+ reference_data: DataFrame,
+ target_data: DataFrame,
+ target_name: str,
+ reference_columns: Index,
+ drift_metrics: list[str],
+ chunk_period: str,
+ metadata: dict[str, Any],
+ ) -> DataFrame:
+ """
+ Calculate univariate drift metrics for a model.
+ Args:
+ model_analysis (ModelAnalysis): Model analysis object
+ reference_data (DataFrame): Reference data
+ target_data (DataFrame): Target data
+ reference_columns (list[str]): Reference columns
+ drift_metrics (list[str]): Drift metrics
+ metadata (dict[str, Any]): Workflow execution metadata
+ """
+
+ config = {
+ 'target': target_name,
+ 'prediction': 'prediction',
+ 'timestamp': 'timestamp',
+ 'features': reference_columns,
+ }
+
+ model_analysis = ModelAnalysis(config=config)
+
+ self.debug(
+ f'Reference data: Size {reference_data.shape} \n{reference_data.head(5).to_string()}',
+ metadata,
+ )
+
+ self.debug(
+ f'Target data: Size {target_data.shape} \n{target_data.head(5).to_string()}', metadata
+ )
+
+ core_labels = self.get_core_labels(metadata, operation_type='detect_univariate_drift')
+ start_time = time.time()
+ try:
+ univariate_drift = model_analysis.detect_univariate_drift(
+ reference_df=reference_data,
+ analysis_df=target_data,
+ features=reference_columns,
+ timestamp_col=config['timestamp'],
+ methods=drift_metrics,
+ chunk_period=chunk_period,
+ )
+ except Exception as e:
+ self.error(f'Error detecting univariate drift: {e}', metadata)
+ 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)
+
+ core_labels = self.get_core_labels(metadata, operation_type='detect_multivariate_drift')
+ start_time = time.time()
+ try:
+ multivariate_drift = model_analysis.detect_multivariate_drift(
+ reference_df=reference_data,
+ analysis_df=target_data,
+ features=reference_columns,
+ timestamp_col=config['timestamp'],
+ chunk_period=chunk_period,
+ )
+ except Exception as e:
+ self.error(f'Error detecting multivariate drift: {e}', metadata)
+ 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)
+
+ start_time = time.time()
+ core_labels = self.get_core_labels(metadata, operation_type='get_drift_metrics_dataframe')
+ try:
+ drift_df = model_analysis.get_drift_metrics_dataframe(
+ univariate_drift=univariate_drift,
+ multivariate_drift=multivariate_drift,
+ )
+ except Exception as e:
+ self.error(f'Error getting drift metrics: {e}', metadata)
+ 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)
+
+ self.debug(
+ f'Drift dataframe: Size {drift_df.shape} \n{drift_df.head(5).to_string()}', metadata
+ )
+
+ return drift_df
+
+ @activity.defn(name='calculate_drift')
+ async def calculate_drift(self, input_data: dict[str, Any]) -> list[dict]:
+ """
+ Calculate drift metrics for a model.
+
+ Args:
+ input_data (dict[str, Any]): Input data containing:
+ - metadata (dict): Workflow execution metadata
+ - model_name (str): Name of the MLFlow model to calculate drift for
+ - reference_data (pd.DataFrame): Reference data for the model
+ - target_data (pd.DataFrame): Target data for calculating drift
+ - target_name (str): Name of the target column
+ - drift_metrics (list[str]): List of drift metrics to calculate
+ """
+ metadata = input_data['metadata']
+ model_name = input_data['model_name']
+ model_id = input_data['model_id']
+ reference_raw_data = input_data['reference_data']
+ target_data = DataFrame(input_data['target_data'])
+ target_name = input_data['target_name']
+ drift_metrics = input_data['drift_metrics']
+ chunk_period = input_data['chunk_period']
+
+ if chunk_period not in ['min', 's']:
+ self.error(f'Invalid chunk period: {chunk_period}', metadata)
+ raise ValueError(f'Invalid chunk period: {chunk_period}, must be "min" or "s"')
+
+ self.info(f'Calculating drift for model {model_name}', metadata)
+
+ target_data = target_data.pivot(index='timestamp', columns='variable', values='value')
+ target_data['timestamp'] = target_data.index
+ target_data['timestamp'] = to_datetime(target_data['timestamp'])
+ target_data['timestamp'] = target_data['timestamp'].dt.strftime(DATETIME_FORMAT)
+ target_data = target_data.reset_index(drop=True)
+ target_data.dropna(inplace=True)
+
+ if reference_raw_data is not None:
+ self.info('Using reference data', metadata)
+ reference_data = DataFrame(reference_raw_data)
+ accurate = True
+ else:
+ # Get 30% first rows of target_data
+ self.warning('Using 30% first rows of target data as reference data', metadata)
+ target_data.sort_values(by='timestamp', ascending=True, inplace=True)
+ reference_data = target_data.head(int(len(target_data) * 0.3))
+ accurate = False
+
+ await self.send_notification_async(
+ metadata=metadata,
+ notification_id='MODEL_METRICS_REFERENCE_DATA_WARNING',
+ message='Using 30% first rows of target data as reference data',
+ block='model_metrics',
+ level=NotificationLevel.WARNING,
+ attachment_content=reference_data.to_csv(),
+ )
+
+ reference_columns = reference_data.drop(
+ columns=[target_name, 'timestamp', 'target', 'prediction'], errors='ignore'
+ ).columns
+
+ try:
+ drift_df = await self.get_drift_metrics(
+ reference_data=reference_data,
+ target_data=target_data,
+ target_name=target_name,
+ reference_columns=reference_columns,
+ drift_metrics=drift_metrics,
+ chunk_period=chunk_period,
+ metadata=metadata,
+ )
+ except Exception as e:
+ self.error(f'Error getting drift metrics: {e}', metadata)
+ await self.send_notification_async(
+ metadata=metadata,
+ notification_id='MODEL_METRICS_GET_DRIFT_METRICS_ERROR',
+ message=f'Error getting drift metrics: {e}',
+ block='model_metrics',
+ level=NotificationLevel.ERROR,
+ attachment_content=traceback.format_exc(),
+ )
+ return []
+
+ if drift_df.empty:
+ self.warning('No drift metrics found', metadata)
+ return []
+
+ # Drop unnecessary columns
+ drift_df.drop(columns=['p_value'], inplace=True)
+
+ # Extract timestamps only until minutes
+ if chunk_period == 'min':
+ target_timestamps = target_data['timestamp'].apply(lambda x: x[:16])
+ else:
+ target_timestamps = target_data['timestamp']
+
+ # Drop rows where timestamp is not in target data, to avoid save drift from reference
+ drift_df = drift_df[drift_df['timestamp'].isin(target_timestamps)]
+
+ if drift_df.empty:
+ self.warning(
+ 'No drift metrics found after dropping rows where timestamp is not in target data',
+ metadata,
+ )
+ return []
+
+ # Rename columns to match database columns
+ drift_df.rename(
+ columns={
+ 'metric': 'method',
+ 'statistic': 'value',
+ },
+ inplace=True,
+ )
+
+ # Drop duplicates
+ drift_df.drop_duplicates(
+ subset=['timestamp', 'method', 'feature'], keep='first', inplace=True
+ )
+
+ drift_df['model_id'] = model_id
+ drift_df['accurate'] = accurate
+
+ drift_df['timestamp'] = to_datetime(drift_df['timestamp'])
+ 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)
+
+ return drift_df.to_dict(orient='records')
+
+ @activity.defn(name='calculate_simple_metrics')
+ async def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]:
+ """
+ Calculate simple metrics for a model. Metrics available are:
+ - rmse
+ - mse
+ - mae
+ - r2
+ - accuracy
+ - precision
+ - recall
+ - f1
+ Args:
+ input_data (dict[str, Any]): Input data containing:
+ - metadata (dict): Workflow execution metadata
+ - model_id (str): ID of the MLFlow model
+ - target_data (pd.DataFrame): Target data for calculating metrics, containing target and prediction columns
+ - metrics (list[str]): List of metrics to calculate
+ Returns:
+ dict[Hashable, Any]: Dictionary containing the calculated metrics
+ """
+
+ metadata = input_data['metadata']
+ model_id = input_data['model_id']
+ target_data = DataFrame(input_data['target_data'])
+ metrics = input_data['metrics']
+ interval_minutes = input_data['interval_minutes']
+
+ data_size = target_data.shape[0]
+
+ output_data = []
+
+ diff = target_data['target'] - target_data['prediction']
+ diff_squared = diff**2
+
+ self.info(f'Calculating simple metrics for model {model_id}: {metrics}', metadata)
+
+ for metric in metrics:
+ if metric == 'rmse':
+ output_data.append({'metric': 'rmse', 'value': np.sqrt(np.mean(diff_squared))})
+ elif metric == 'mse':
+ output_data.append({'metric': 'mse', 'value': np.mean(diff_squared)})
+ elif metric == 'mae':
+ output_data.append({'metric': 'mae', 'value': np.mean(np.abs(diff))})
+ elif metric == 'r2':
+ y_true = target_data['target']
+ y_mean = np.mean(y_true)
+
+ ss_res = np.sum(diff_squared)
+ ss_tot = np.sum((y_true - y_mean) ** 2)
+
+ # Evita divisão por zero
+ if ss_tot == 0:
+ r2_score = 0.0
+ else:
+ r2_score = 1 - (ss_res / ss_tot)
+
+ output_data.append({'metric': 'r2', 'value': r2_score})
+
+ data = DataFrame(output_data)
+ data['model_id'] = model_id
+ data['timestamp'] = target_data['timestamp'].max()
+ 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
+ )
+
+ return data.to_dict(orient='records')
diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py
index aa93ca5..c4abe95 100644
--- a/laborious/activities/opc.py
+++ b/laborious/activities/opc.py
@@ -2,6 +2,7 @@ from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
import traceback
+ from collections.abc import Hashable
from typing import Any
from pandas import DataFrame
@@ -275,7 +276,7 @@ class OPC(SientiaMonitoring):
@activity.defn(name='write_opc_data')
async def write_opc_data(
self, input_data: dict[str, Any]
- ) -> tuple[dict[Any, Any], dict[str, dict[str, float | None]]]:
+ ) -> tuple[dict[Hashable, Any], dict[str, dict[str, float | None]]]:
"""
Write prediction and confidence data to OPC servers. The two writing
operations are optional and independent of each other.
@@ -324,7 +325,7 @@ class OPC(SientiaMonitoring):
def process_confidence(
self, data: DataFrame, success: bool, metadata: dict[str, Any]
- ) -> dict[Any, Any]:
+ ) -> dict[Hashable, Any]:
"""
Process prediction confidence based on OPC write operation success.
diff --git a/laborious/metrics.py b/laborious/metrics.py
index ec87867..be5604d 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,
+)
diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py
index 1414713..eaab16c 100644
--- a/laborious/utils/repository/model_repository.py
+++ b/laborious/utils/repository/model_repository.py
@@ -20,6 +20,7 @@ import threading
import time
import traceback
from datetime import datetime, timedelta
+from io import StringIO
from os import environ, makedirs, path
from shutil import rmtree
from typing import Any, Literal, overload
@@ -215,6 +216,26 @@ class MLFlowRepository(SientiaMonitoring):
run_info = mlflow.get_run(run_id)
return run_info.data.params
+ def check_artifact_exists(
+ self, run_id: str, artifact_path: str, metadata: dict[str, Any]
+ ) -> bool:
+ """
+ Check if an artifact exists in the MLflow Model Registry.
+
+ Args:
+ run_id (str): Run identifier to inspect.
+ artifact_path (str): Path to the artifact to check.
+
+ Returns:
+ bool: True if the artifact exists, False otherwise.
+ """
+ artifacts = self.client.list_artifacts(run_id)
+
+ self.debug(f'Artifacts of {run_id}: \n{artifacts}', metadata)
+ self.debug(f'Looking for artifact {artifact_path} in {run_id}', metadata)
+
+ return any(artifact.path == artifact_path for artifact in artifacts)
+
"""
Functions related to download and load models
"""
@@ -259,6 +280,45 @@ class MLFlowRepository(SientiaMonitoring):
return artifacts
+ async def load_artifact_dataframe(
+ self, model_name: str, artifact_path: str, metadata: dict[str, Any]
+ ) -> pd.DataFrame | None:
+ """
+ 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')
+ core_labels = self.get_core_labels(metadata, operation_type='load_text')
+
+ if not self.check_artifact_exists(run_id, artifact_path, metadata):
+ return None
+
+ artifact_path = path.join('runs:/', run_id, artifact_path)
+
+ 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)
+
+ self.debug(f'Content of {run_id}/{artifact_path}: \n{content}', metadata)
+
+ 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:
@@ -644,6 +704,47 @@ class MLFlowRepository(SientiaMonitoring):
Functions related to model retraining
"""
+ def get_prediction_data(
+ self,
+ prediction_model: Any,
+ retrain_dataset: pd.DataFrame,
+ target_name: str,
+ predict_flavor: str,
+ ) -> pd.DataFrame:
+ """
+ Get prediction data from prediction model.
+ """
+ input_index = retrain_dataset.index
+
+ if predict_flavor == 'pyfunc':
+ prediction_data = prediction_model.predict({}, retrain_dataset)
+ else:
+ prediction_data = prediction_model.predict(retrain_dataset)
+
+ if isinstance(prediction_data, pd.DataFrame):
+ prediction_data.columns = pd.Index(['prediction'])
+
+ else:
+ prediction_data = pd.DataFrame(prediction_data, columns=['prediction'])
+
+ prediction_data.index = input_index
+
+ # Merge prediction data with retrain_dataset on index
+ prediction_data = pd.merge( # NOSONAR
+ retrain_dataset, prediction_data, left_index=True, right_index=True, how='left'
+ )
+
+ # Rename column "target_name" to "target"
+ prediction_data.rename(columns={target_name: 'target'}, inplace=True)
+
+ prediction_data['timestamp'] = prediction_data.index
+
+ prediction_data.reset_index(drop=True, inplace=True)
+
+ prediction_data = prediction_data.sort_values(by='timestamp', ascending=True)
+
+ return prediction_data
+
async def fit_models(
self,
model_name: str,
@@ -653,7 +754,7 @@ class MLFlowRepository(SientiaMonitoring):
transform_flavor: str = 'sklearn',
predict_flavor: str = 'sklearn',
target_name: str | None = None,
- ) -> dict[str, dict[str, Any]]:
+ ) -> dict[str, Any]:
"""
Prepare models and data for a retraining run.
@@ -763,6 +864,11 @@ class MLFlowRepository(SientiaMonitoring):
prediction_model.fit(retrain_dataset)
+ # get prediction data
+ prediction_data = self.get_prediction_data(
+ prediction_model, retrain_dataset, target_name, predict_flavor
+ )
+
self.info(f'Model experiment creation completed successfully for {model_name}', metadata)
retrain_data = {
@@ -771,6 +877,7 @@ class MLFlowRepository(SientiaMonitoring):
'artifact_path': prediction_artifact_path,
},
'data_model': {'model': data_model, 'artifact_path': data_artifact_path},
+ 'prediction_data': prediction_data,
}
return retrain_data
@@ -849,6 +956,7 @@ class MLFlowRepository(SientiaMonitoring):
prediction_model = retrain_data['prediction_model']
data_model = retrain_data['data_model']
+ prediction_data = retrain_data['prediction_data']
model_temp_path = path.join(ARTIFACTS_PATH, model_name)
@@ -872,10 +980,12 @@ class MLFlowRepository(SientiaMonitoring):
self.debug(f'Attributes: {retrain_params}', metadata)
data_path = f'{model_temp_path}/retrain_data.csv'
+ prediction_data_path = f'{model_temp_path}/evaluation_data.csv'
makedirs(model_temp_path, exist_ok=True)
- data.to_csv(data_path, index=True)
+ data.to_csv(data_path, index=False)
+ prediction_data.to_csv(prediction_data_path, index=False)
self.info(
f'Starting model upload for {experiment_name} with run name {current_run_name}',
@@ -909,6 +1019,7 @@ class MLFlowRepository(SientiaMonitoring):
# log the data raw
mlflow.log_artifact(data_path)
+ mlflow.log_artifact(prediction_data_path)
except Exception as e:
await self.emit_metric(metric_object=metrics.MODEL_WRITE_ERROR_COUNT, tags=core_labels)
diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py
index 5faf004..33384b0 100644
--- a/laborious/worker/worker.py
+++ b/laborious/worker/worker.py
@@ -5,12 +5,15 @@ This module provides the main worker implementation for the Sientia DataOps Labo
It orchestrates Temporal workers, manages task queues, and handles the lifecycle of
prediction and retraining workflows.
-The worker supports two main task queues:
-- predictions_batch-queue: Handles batch prediction workflows
+The worker supports multiple task queues:
+- predictions_batch-queue: Handles batch prediction workflows (heavy workload)
- minimal_retrain-queue: Handles model retraining workflows
+- drift-queue: Handles drift detection workflows
+- simple_metrics-queue: Handles simple metrics calculation workflows
Key Features:
-- Automatic scaling with PollerBehaviorAutoscaling
+- Resource-based scaling with WorkerTuner (CPU and memory aware)
+- Automatic polling scaling with PollerBehaviorAutoscaling
- Prometheus metrics integration
- Comprehensive error handling and logging
- Graceful shutdown with cleanup
@@ -23,16 +26,37 @@ Environment Variables:
- HTTP_METRICS_PORT: Prometheus metrics server port (default: 9090)
- HTTP_SDK_METRICS_PORT: Temporal SDK metrics port (default: 9091)
- PROJECT_NAME: Project name for notifications (default: laborious)
+
+Tuner Configuration (Resource-based scaling):
+- TUNER_TARGET_MEMORY_USAGE: Target memory usage (0.0-1.0, default: 0.75)
+- TUNER_TARGET_CPU_USAGE: Target CPU usage (0.0-1.0, default: 0.80)
+- TUNER_WORKFLOW_MIN_SLOTS: Minimum workflow slots (default: 5)
+- TUNER_WORKFLOW_MAX_SLOTS: Maximum workflow slots (default: 50)
+- TUNER_ACTIVITY_MIN_SLOTS: Minimum activity slots (default: 5)
+- TUNER_ACTIVITY_MAX_SLOTS: Maximum activity slots (default: 50)
+- TUNER_WORKFLOW_RAMP_THROTTLE_MS: Workflow ramp throttle in ms (default: 100)
+- TUNER_ACTIVITY_RAMP_THROTTLE_MS: Activity ramp throttle in ms (default: 50)
+
+Poller Configuration:
+- POLLER_MINIMUM: Minimum number of pollers (default: 1)
+- POLLER_MAXIMUM: Maximum number of pollers (default: 10)
+- POLLER_INITIAL: Initial number of pollers (default: 2)
"""
+
from temporalio import client, workflow
from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
-from temporalio.worker import PollerBehaviorAutoscaling, Worker
+from temporalio.worker import (
+ PollerBehaviorAutoscaling,
+ ResourceBasedSlotConfig,
+ Worker,
+ WorkerTuner,
+)
with workflow.unsafe.imports_passed_through():
import asyncio
- import os
import sys
+ from datetime import timedelta
from prometheus_client import start_http_server
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -47,17 +71,63 @@ with workflow.unsafe.imports_passed_through():
build_opc_config,
build_postgres_config,
)
+ from laborious.workflows.drift import Drift
from laborious.workflows.minimal_retrain import MinimalRetrain
from laborious.workflows.predictions_batch import PredictionsBatch
+ from laborious.workflows.simple_metrics import SimpleMetrics
from laborious.workflows.sub_workflows.format_and_export_prediction import (
FormatAndExportPrediction,
)
from laborious.workflows.sub_workflows.prediction_process import PredictionProcess
+ import os
POD_ID = os.getenv('POD_ID')
SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091'))
+def create_resource_tuner() -> WorkerTuner:
+ """Create a resource-based tuner from environment variables."""
+ target_memory = float(os.getenv('TUNER_TARGET_MEMORY_USAGE', '0.75'))
+ target_cpu = float(os.getenv('TUNER_TARGET_CPU_USAGE', '0.50'))
+ workflow_min = int(os.getenv('TUNER_WORKFLOW_MIN_SLOTS', '5'))
+ workflow_max = int(os.getenv('TUNER_WORKFLOW_MAX_SLOTS', '50'))
+ activity_min = int(os.getenv('TUNER_ACTIVITY_MIN_SLOTS', '5'))
+ activity_max = int(os.getenv('TUNER_ACTIVITY_MAX_SLOTS', '50'))
+ local_activity_min = int(os.getenv('TUNER_LOCAL_ACTIVITY_MIN_SLOTS', '1'))
+ local_activity_max = int(os.getenv('TUNER_LOCAL_ACTIVITY_MAX_SLOTS', '30'))
+ workflow_ramp = int(os.getenv('TUNER_WORKFLOW_RAMP_THROTTLE_MS', '100'))
+ activity_ramp = int(os.getenv('TUNER_ACTIVITY_RAMP_THROTTLE_MS', '50'))
+ local_activity_ramp = int(os.getenv('TUNER_LOCAL_ACTIVITY_RAMP_THROTTLE_MS', '50'))
+
+ return WorkerTuner.create_resource_based(
+ target_memory_usage=target_memory,
+ target_cpu_usage=target_cpu,
+ workflow_config=ResourceBasedSlotConfig(
+ minimum_slots=workflow_min,
+ maximum_slots=workflow_max,
+ ramp_throttle=timedelta(milliseconds=workflow_ramp),
+ ),
+ activity_config=ResourceBasedSlotConfig(
+ minimum_slots=activity_min,
+ maximum_slots=activity_max,
+ ramp_throttle=timedelta(milliseconds=activity_ramp),
+ ),
+ local_activity_config=ResourceBasedSlotConfig(
+ minimum_slots=local_activity_min,
+ maximum_slots=local_activity_max,
+ ramp_throttle=timedelta(milliseconds=local_activity_ramp),
+ ),
+ )
+
+
+def create_poller_behavior() -> PollerBehaviorAutoscaling:
+ """Create poller behavior from environment variables."""
+ minimum = int(os.getenv('POLLER_MINIMUM', '1'))
+ maximum = int(os.getenv('POLLER_MAXIMUM', '10'))
+ initial = int(os.getenv('POLLER_INITIAL', '2'))
+ return PollerBehaviorAutoscaling(minimum=minimum, maximum=maximum, initial=initial)
+
+
async def main():
"""
Main entry point for the Laborious worker application.
@@ -136,6 +206,9 @@ async def main():
logger.custom_info('Starting Workers...', metadata)
+ tuner = create_resource_tuner()
+ poller = create_poller_behavior()
+
workers = [
Worker(
temporal_client,
@@ -149,12 +222,39 @@ async def main():
activities.format_retrain_report,
activities.export_data_to_postgres,
],
- max_concurrent_workflow_tasks=50,
- max_concurrent_activities=50,
- max_concurrent_local_activities=50,
+ tuner=tuner,
max_cached_workflows=2,
- workflow_task_poller_behavior=PollerBehaviorAutoscaling(),
- activity_task_poller_behavior=PollerBehaviorAutoscaling(),
+ workflow_task_poller_behavior=poller,
+ activity_task_poller_behavior=poller,
+ ),
+ Worker(
+ temporal_client,
+ task_queue='drift-queue',
+ workflows=[Drift],
+ activities=[
+ activities.load_custom_query,
+ activities.get_reference_data,
+ activities.calculate_drift,
+ activities.export_data_to_postgres,
+ ],
+ tuner=tuner,
+ max_cached_workflows=2,
+ workflow_task_poller_behavior=poller,
+ activity_task_poller_behavior=poller,
+ ),
+ Worker(
+ temporal_client,
+ task_queue='simple_metrics-queue',
+ workflows=[SimpleMetrics],
+ activities=[
+ activities.load_custom_query,
+ activities.calculate_simple_metrics,
+ activities.export_data_to_postgres,
+ ],
+ tuner=tuner,
+ max_cached_workflows=2,
+ workflow_task_poller_behavior=poller,
+ activity_task_poller_behavior=poller,
),
Worker(
temporal_client,
@@ -168,6 +268,7 @@ async def main():
activities.input_gate,
activities.mlflow_response_gate,
activities.mlflow_content_gate,
+ activities.format_transformed_data,
activities.format_prediction,
activities.format_default_prediction,
activities.get_last_timestamp,
@@ -179,12 +280,10 @@ async def main():
activities.export_data_to_postgres,
activities.write_metrics,
],
- max_concurrent_workflow_tasks=50,
- max_concurrent_activities=50,
- max_concurrent_local_activities=50,
+ tuner=tuner,
max_cached_workflows=200,
- workflow_task_poller_behavior=PollerBehaviorAutoscaling(),
- activity_task_poller_behavior=PollerBehaviorAutoscaling(),
+ workflow_task_poller_behavior=poller,
+ activity_task_poller_behavior=poller,
),
]
diff --git a/laborious/workflows/drift.py b/laborious/workflows/drift.py
new file mode 100644
index 0000000..769c613
--- /dev/null
+++ b/laborious/workflows/drift.py
@@ -0,0 +1,107 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+ from datetime import timedelta
+ from typing import Any
+
+ from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
+ from sientia_do.temporal.policies import retry_policy
+
+ from laborious.activities.activities import Activities
+
+
+@workflow.defn(name='drift')
+class Drift:
+ @workflow.run
+ async def run(self, input_data: dict[str, Any]):
+ """
+ Execute the drift workflow.
+
+ This method orchestrates the complete drift process by:
+ 1. Loading data using the provided custom SQL query
+ 2. Preparing prediction configuration and filters
+ 3. Delegating to the PredictionProcess workflow for ML operations
+ """
+
+ metadata = {
+ 'metadata': {
+ 'schedule_name': input_data['schedule_name'],
+ 'model_name': input_data['model_name'],
+ 'model_id': input_data['model_id'],
+ 'workflow_name': 'drift',
+ }
+ }
+
+ print(f'Input data: {input_data}', metadata)
+
+ model_config = input_data['model_config']
+ target_name = model_config['target']
+
+ gathering_query = f"""
+ SELECT *
+ FROM "{input_data['schema']}"."{input_data['source_table_name']}"
+ WHERE
+ model_id = '{input_data['model_id']}' AND
+ timestamp > NOW() - INTERVAL '{input_data['interval']} minutes'
+ ORDER BY timestamp ASC
+ """
+
+ target_data_handler = workflow.start_local_activity_method(
+ Activities.load_custom_query,
+ {
+ **metadata,
+ 'query': gathering_query,
+ 'datetime_columns': ['timestamp', 'created_at'],
+ 'orient': 'records',
+ },
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=300),
+ )
+
+ reference_data_handler = workflow.start_local_activity_method(
+ Activities.get_reference_data,
+ {**metadata, 'model_name': input_data['model_name']},
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=300),
+ )
+
+ target_data = await target_data_handler
+ reference_data = await reference_data_handler
+
+ if not target_data:
+ return
+
+ drift_data = await workflow.execute_local_activity_method(
+ Activities.calculate_drift,
+ {
+ **metadata,
+ 'target_data': target_data,
+ 'reference_data': reference_data,
+ 'model_name': input_data['model_name'],
+ 'model_id': input_data['model_id'],
+ 'target_name': target_name,
+ 'drift_metrics': input_data.get(
+ 'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
+ ),
+ 'chunk_period': input_data.get('chunk_period', 'min'),
+ },
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=300),
+ )
+
+ if drift_data:
+ await workflow.execute_activity_method(
+ Activities.export_data_to_postgres,
+ {
+ **metadata,
+ 'data': drift_data,
+ 'schema': input_data['schema'],
+ 'table_name': input_data['target_table_name'],
+ 'timestamp_conversion': {
+ 'column': 'timestamp',
+ 'format': DATETIME_FORMAT_WITH_TZ,
+ },
+ },
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=300),
+ )
diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py
index afd7c86..6c5f770 100644
--- a/laborious/workflows/predictions_batch.py
+++ b/laborious/workflows/predictions_batch.py
@@ -98,6 +98,7 @@ class PredictionsBatch:
'data': data,
'schema': input_data['schema'],
'table_name': input_data['table_name'],
+ 'transform_table_name': input_data['transform_table_name'],
'model_id': input_data['model_id'],
'model_name': input_data['model_name'],
'input_filters': input_data.get('input_filters', {'EMPTY_DATA': {'POLICY': 'STOP'}}),
diff --git a/laborious/workflows/simple_metrics.py b/laborious/workflows/simple_metrics.py
new file mode 100644
index 0000000..438b98d
--- /dev/null
+++ b/laborious/workflows/simple_metrics.py
@@ -0,0 +1,95 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+ from datetime import timedelta
+ from typing import Any
+
+ from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
+ from sientia_do.temporal.policies import retry_policy
+
+ from laborious.activities.activities import Activities
+
+
+@workflow.defn(name='simple_metrics')
+class SimpleMetrics:
+ @workflow.run
+ async def run(self, input_data: dict[str, Any]):
+ """
+ Execute the simple metrics workflow.
+ """
+ metadata = {
+ 'metadata': {
+ 'model_id': input_data['model_id'],
+ 'model_name': input_data['model_name'],
+ 'workflow_name': 'simple_metrics',
+ 'schedule_name': input_data['schedule_name'],
+ }
+ }
+
+ model_id = input_data['model_id']
+ interval_minutes = input_data['interval_minutes']
+
+ model_config = input_data['model_config']
+ target_name = model_config['target']
+
+ query = f"""
+ select p."timestamp", p.prediction, ld.value as "target"
+ from "{input_data['schema']}"."{input_data['predictions_table_name']}" p
+ inner join "{input_data['schema']}"."{input_data['data_table_name']}" ld
+ on p."timestamp" = ld."timestamp"
+ where
+ p.model_id = '{model_id}' and
+ p.prediction is not null and
+ ld.variable = '{target_name}' and
+ ld.value is not null and
+ p."timestamp" >= NOW() - INTERVAL '{interval_minutes} minutes'
+ order by
+ p."timestamp" desc;
+ """
+
+ target_data = await workflow.execute_local_activity_method(
+ Activities.load_custom_query,
+ {
+ **metadata,
+ 'query': query,
+ 'datetime_columns': ['timestamp'],
+ 'orient': 'records',
+ },
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=300),
+ )
+
+ if not target_data:
+ return
+
+ simple_metrics = await workflow.execute_local_activity_method(
+ Activities.calculate_simple_metrics,
+ {
+ **metadata,
+ 'model_id': model_id,
+ 'target_data': target_data,
+ 'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+ 'interval_minutes': interval_minutes,
+ },
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=300),
+ )
+
+ if not simple_metrics:
+ return
+
+ await workflow.execute_activity_method(
+ Activities.export_data_to_postgres,
+ {
+ **metadata,
+ 'data': simple_metrics,
+ 'schema': input_data['schema'],
+ 'table_name': input_data['target_table_name'],
+ 'timestamp_conversion': {
+ 'column': 'timestamp',
+ 'format': DATETIME_FORMAT_WITH_TZ,
+ },
+ },
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=300),
+ )
diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py
index 8d8c9da..e50c2cf 100644
--- a/laborious/workflows/sub_workflows/format_and_export_prediction.py
+++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py
@@ -50,29 +50,45 @@ class FormatAndExportPrediction:
Args:
input_data: Complete configuration for the export workflow
Required keys:
+ - metadata (dict): Workflow execution metadata
- path_flag (str | None): Decision path flag for formatting strategy
+ - None: Normal prediction path with full formatting
+ - Any other value: Default prediction path for error conditions
- data (dict[str, Any]): Prediction data to format and export
- prediction_confidence (float): Confidence score for the prediction
- timestamp (str): ISO-formatted timestamp for the prediction
- model_id (int): Unique identifier for the ML model
- model_name (str): Name of the ML model
- - model_retention (str): Model retention policy configuration
- - comment (str): Operational comment or error description
- schema (str): Database schema for data storage
- table_name (str): Target table for data persistence
- opc_output_config (dict[str, Any]): OPC server export configuration
- - prediction_store_policy (str, optional): Data retention policy
+ Optional keys:
+ - transformed_data (dict[str, Any]): Transformed data to export separately
+ Only processed when path_flag is None
+ - transform_table_name (str): Target table for transformed data export
+ Required if transformed_data is provided
+ - prediction_store_policy (str): Data retention policy (e.g., 'lts:1', 'erl:2')
+ Required when path_flag is None
+ - comment (str): Operational comment or error description
+ Required when path_flag is not None
Returns:
- bool: True if the workflow completes successfully, False otherwise
+ None: The workflow completes successfully when all export operations finish
+
+ Note:
+ When transformed_data is provided and path_flag is None, the workflow will:
+ 1. Format the transformed data using format_transformed_data
+ 2. Export it to a separate table (transform_table_name) asynchronously
+ 3. Wait for both prediction and transformed data exports to complete
"""
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:
- # proceed with formatting and exporting
+ # Normal prediction path: format prediction data with full metadata
prediction = await workflow.execute_local_activity_method(
Activities.format_prediction,
{
@@ -87,8 +103,40 @@ class FormatAndExportPrediction:
start_to_close_timeout=timedelta(seconds=60),
)
+ # Optionally format and export transformed data to separate table
+ if transformed_data is not None:
+ transformed = await workflow.execute_local_activity_method(
+ Activities.format_transformed_data,
+ {
+ **metadata,
+ 'data': transformed_data,
+ 'model_id': input_data['model_id'],
+ },
+ retry_policy=retry_policy,
+ start_to_close_timeout=timedelta(seconds=60),
+ )
+
+ write_transformed_handler = workflow.start_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
+ # Error path: create default prediction with error indicators
prediction = await workflow.execute_local_activity_method(
Activities.format_default_prediction,
{
@@ -102,6 +150,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 +165,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 +178,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..73f0227 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,
@@ -207,6 +209,7 @@ class PredictionProcess:
'opc_output_config': input_data['opc_output_config'],
'schema': input_data['schema'],
'table_name': input_data['table_name'],
+ 'transform_table_name': input_data['transform_table_name'],
'comment': comment,
'prediction_store_policy': input_data['prediction_store_policy'],
},
@@ -248,6 +251,7 @@ class PredictionProcess:
schema = input_data['schema']
table_name = input_data['table_name']
+ transform_table_name = input_data['transform_table_name']
model_id = input_data['model_id']
model_name = input_data['model_name']
model_config = input_data.get('model_config', {})
@@ -287,6 +291,7 @@ class PredictionProcess:
'model_config': model_config,
'schema': schema,
'table_name': table_name,
+ 'transform_table_name': transform_table_name,
'comment': comment,
'opc_output_config': input_data['opc_output_config'],
'prediction_store_policy': input_data['prediction_store_policy'],
diff --git a/pyproject.toml b/pyproject.toml
index 12e06d4..0f912c2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -48,6 +48,7 @@ ignore = [
"S101", # use of assert (needed for tests)
"S105", # possible hardcoded password (false positives)
"S106", # possible hardcoded password (false positives)
+ "S608", # potential sql injection (false positives)
"N802", # function name should be lowercase (temporal decorators)
"N806", # variable in function should be lowercase
]
@@ -152,4 +153,4 @@ directory = "htmlcov"
[tool.bandit]
exclude_dirs = ["tests", "venv", ".venv"]
-skips = ["B101", "B601"] # Skip assert and shell injection in controlled environments
\ No newline at end of file
+skips = ["B101", "B601", "B608"] # Skip assert, shell injection, and SQL injection (false positives)
\ No newline at end of file
diff --git a/requirements-light.txt b/requirements-light.txt
index e41ba53..8831822 100644
--- a/requirements-light.txt
+++ b/requirements-light.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.6.1
prometheus-client
botocore
boto3
diff --git a/requirements.txt b/requirements.txt
index 12feb2e..3921162 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,8 +3,8 @@ 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-mlops-library.git@0.39.0
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1
+git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.40.6
prometheus-client
botocore
boto3
diff --git a/tests.ipynb b/tests.ipynb
index 3e4b61a..b8c1b5c 100644
--- a/tests.ipynb
+++ b/tests.ipynb
@@ -487,6 +487,283 @@
"except Exception as e:\n",
" print(e)\n"
]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 7,
+ "id": "771ab4ee",
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "text/html": [
+ "\n",
+ "\n",
+ "
\n",
+ " \n",
+ " \n",
+ " | \n",
+ " a | \n",
+ " b | \n",
+ " target | \n",
+ " prediction | \n",
+ " timestamp | \n",
+ "
\n",
+ " \n",
+ " \n",
+ " \n",
+ " | 0 | \n",
+ " 1 | \n",
+ " 4 | \n",
+ " 7 | \n",
+ " 1 | \n",
+ " 2025-01-01 | \n",
+ "
\n",
+ " \n",
+ " | 1 | \n",
+ " 2 | \n",
+ " 5 | \n",
+ " 8 | \n",
+ " 2 | \n",
+ " 2025-01-02 | \n",
+ "
\n",
+ " \n",
+ " | 2 | \n",
+ " 3 | \n",
+ " 6 | \n",
+ " 9 | \n",
+ " 3 | \n",
+ " 2025-01-03 | \n",
+ "
\n",
+ " \n",
+ "
\n",
+ "
\n",
+ "\n",
+ "
\n",
+ " \n",
+ " \n",
+ " | \n",
+ " a | \n",
+ " b | \n",
+ "
\n",
+ " \n",
+ " \n",
+ " \n",
+ " | 2025-01-01 | \n",
+ " 1 | \n",
+ " 4 | \n",
+ "
\n",
+ " \n",
+ " | 2025-01-02 | \n",
+ " 2 | \n",
+ " 5 | \n",
+ "
\n",
+ " \n",
+ " | 2025-01-03 | \n",
+ " 3 | \n",
+ " 6 | \n",
+ "
\n",
+ " \n",
+ "
\n",
+ "
\n",
+ "\n",
+ "
\n",
+ " \n",
+ " \n",
+ " | \n",
+ " 0 | \n",
+ " 1 | \n",
+ " 2 | \n",
+ "
\n",
+ " \n",
+ " \n",
+ " \n",
+ " | index | \n",
+ " 2025-01-01 | \n",
+ " 2025-01-02 | \n",
+ " 2025-01-03 | \n",
+ "
\n",
+ " \n",
+ " | columns | \n",
+ " a | \n",
+ " b | \n",
+ " None | \n",
+ "
\n",
+ " \n",
+ " | data | \n",
+ " [1, 4] | \n",
+ " [2, 5] | \n",
+ " [3, 6] | \n",
+ "
\n",
+ " \n",
+ "
\n",
+ "