SIENTIAPDE-1222

Update .gitignore to include 'tmp/' directory and ensure '.env' is listed - Added 'tmp/' to the .gitignore file to prevent temporary files from being tracked. - Confirmed that '.env' is included to avoid committing sensitive environment variables.
SIENTIAPDE-1222
2025-09-24 10:33:18 -03:00 · 2025-09-24 10:32:59 -03:00 · 2025-09-24 10:30:23 -03:00 · 2025-09-24 09:59:26 -03:00 · 2025-09-17 16:02:54 -03:00 · 2025-09-17 11:15:05 -03:00
19 changed files with 957 additions and 641 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -43,4 +43,6 @@ git_key*

 git_log

-.env
+.env
+
+tmp/
--- a/README.md
+++ b/README.md
@@ -66,6 +66,7 @@ The Laborious system uses a Temporal-based workflow architecture with clear sepa
  - Health check endpoints for Kubernetes liveness/readiness probes
  - Graceful shutdown with cleanup procedures
  - Multi-instance deployment support
+  - Two dedicated task queues: `predictions_batch-queue` and `minimal_retrain-queue`

 #### **Workflows (`laborious/workflows/`)**
 - **PredictionsBatch**: Main entry point for batch prediction pipelines
@@ -79,25 +80,32 @@ The Laborious system uses a Temporal-based workflow architecture with clear sepa
  - Configurable timeout and retry strategies

 #### **Activities (`laborious/activities/`)**
+- **Activities**: Main activity orchestrator combining all functionality through multiple inheritance
 - **Gates**: Data quality validation and filtering mechanisms
 - **MLFlow**: Model transformation and prediction operations
 - **OPC**: Real-time data export to industrial OPC servers
- **Activities**: Main activity orchestrator and coordination
 - **Key Features**:
+  - Multiple inheritance pattern for unified activity interface
  - Configurable filter policies and validation rules
-  - MLFlow model serving integration
+  - MLFlow model serving integration with configurable flavors
  - OPC UA client with certificate-based authentication
-  - Comprehensive error handling and notification
+  - Comprehensive error handling and notification integration
+  - Support for multiple OPC servers with independent configurations

 #### **Data Services (`laborious/utils/`)**
- **Connectors**: Database and external service configuration management
+- **Connectors Config**: Environment variable-based configuration management
 - **Repository**: Data access layer for MLFlow and OPC operations
+  - `model_repository.py`: MLFlow model operations and retraining
+  - `opc_repository.py`: OPC server communication and data writing
 - **Filters**: Data quality validation and MLFlow response filtering
+  - `conditional_filters.py`: Input data validation filters
+  - `mlflow_filters.py`: MLFlow API response validation filters
 - **Key Features**:
-  - Environment variable-based configuration
+  - Environment variable-based configuration with sensible defaults
  - Connection pool management and optimization
  - Security credential management
  - Configuration validation and error handling
+  - Support for multiple OPC servers and MLFlow model flavors

 ### Data Flow Architecture

@@ -506,23 +514,32 @@ pytest tests/workflow/test_predictions_batch.py

 ## 📊 Monitoring and Metrics

-The Laborious system exposes comprehensive Prometheus metrics:
+The Laborious system exposes comprehensive Prometheus metrics for operational visibility and performance monitoring:

-### Application Metrics
+### Application Health Metrics
 - `app_up`: Application health status (1=healthy, 0=unhealthy)
- `laborious_predictions_written_count`: Prediction export operation count
- `laborious_prediction_confidence_monitor`: Prediction confidence monitoring
- `laborious_prediction_response_time_monitor`: Prediction response time monitoring
+  - Labels: `pod_id`

-### MLFlow Metrics
- Model transformation and prediction success rates
- API response times and error rates
- Model retention and versioning metrics
+### Prediction Operation Metrics
+- `laborious_predictions_written_count`: Counter for successful prediction exports
+  - Labels: `pod_id`, `model_name`, `pipeline_name`
+- `laborious_prediction_confidence_monitor`: Gauge for current prediction confidence levels
+  - Labels: `pod_id`, `model_name`, `pipeline_name`
+- `laborious_prediction_response_time_monitor`: Histogram for prediction response times
+  - Labels: `pod_id`, `model_name`, `pipeline_name`
+  - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]

-### Export Metrics
- PostgreSQL export operation counts and response times
- OPC server write operations and performance
- Data quality filter pass/fail rates
+### OPC Export Metrics
+- `laborious_prediction_opc_writing_count`: Counter for OPC server write operations
+  - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id`
+- `laborious_prediction_opc_writing_response_time_monitor`: Histogram for OPC write response times
+  - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id`
+  - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]
+
+### Data Quality Metrics
+- Filter pass/fail rates through notification system
+- MLFlow API response validation metrics
+- Data quality gate performance tracking

 ## ⚙️ Configuration

@@ -537,14 +554,33 @@ The Laborious system exposes comprehensive Prometheus metrics:
 | `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes |
 | `POSTGRES_PASSWORD` | PostgreSQL password | `sientia` | Yes |
 | `POSTGRES_DBNAME` | PostgreSQL database | `sientia` | Yes |
-| `MLFLOW_HOST` | MLFlow server hostname | `localhost` | Yes |
-| `MLFLOW_PORT` | MLFlow server port | `5000` | Yes |
-| `MLFLOW_USERNAME` | MLFlow username | `admin` | Yes |
-| `MLFLOW_PASSWORD` | MLFlow password | `admin` | Yes |
+| `POSTGRES_MIN_CONNECTIONS` | Minimum PostgreSQL connections | `5` | No |
+| `POSTGRES_MAX_CONNECTIONS` | Maximum PostgreSQL connections | `20` | No |
+| `MLFLOW_HOST` | MLFlow server hostname | `http://localhost` | Yes |
+| `MLFLOW_PORT` | MLFlow server port | `5080` | Yes |
+| `MLFLOW_USERNAME` | MLFlow username | `aignosi` | Yes |
+| `MLFLOW_PASSWORD` | MLFlow password | `aignosi` | Yes |
 | `OPC_CONFIG` | OPC server configuration (JSON) | `{}` | No |
-| `MONGODB_URL` | MongoDB connection URI | `localhost:27017` | Yes |
+| `OPC_ID` | OPC server identifier | `1` | No |
+| `OPC_URL` | OPC server URL | `opc.tcp://localhost:4840` | No |
+| `OPC_SERVER_URI` | OPC server URI | `opc.tcp://localhost:4840` | No |
+| `OPC_CERT_PATH` | OPC client certificate path | `None` | No |
+| `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No |
+| `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No |
+| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No |
+| `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes |
+| `MONGODB_USERNAME` | MongoDB username | `root` | Yes |
+| `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes |
+| `MONGODB_DATABASE_NAME` | MongoDB database name | `sientia` | Yes |
+| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index hours | `1` | No |
+| `LOG_LEVEL` | Application log level | `INFO` | No |
+| `PROJECT_NAME` | Project name for metrics | `laborious` | No |
 | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
 | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No |
+| `POD_ID` | Kubernetes pod identifier | `None` | No |
+
+
+

 ### OPC Configuration

@@ -586,74 +622,88 @@ For single OPC server, use individual environment variables:

 MongoDB pipeline configuration:

-#### Predictions Batch Workflow
+#### Predictions Batch Workflow configuration sample
+
+This is the configuration for the Predictions Batch Workflow, to be inserted into the MongoDB pipeline collection.
+
 ```json
 {
  "schedule_name": "laborious-orchestrated-pipeline",
  "model_id": "1",
  "workflow_type": "predictions_batch",
-  "frequency": "30s", # Workflow execution frequency
-  "max_retry_policy": 1, # Maximum number of retries for the workflow
+  "frequency": "30s",
+  "max_retry_policy": 1,
  "query": "select * from sientia_data.laborious_data where model_id = 1 and \"timestamp\" > NOW() - INTERVAL '5 minutes' order by \"timestamp\" desc limit 30;",
-  "retention_time": 60, # Retention time for models in minutes
  "write_tags": [
    {
-      "server_id": "1",
-      "type": "prediction", # Type of tag to write, can be prediction or confidence
+      "server_id": "server1",
+      "type": "prediction",
      "addr": "ns=2;i=5",
      "data_type": "double"
    },
    {
-      "server_id": "1",
+      "server_id": "server1",
      "type": "confidence",
-      "addr": "ns=2;i=5",
+      "addr": "ns=2;i=6",
      "data_type": "double"
    }
  ],
-  "input_filters": [
-    {
-      "filter_name": "EMPTY_DATA", # Required filter
-      "policy": "STOP"
-    },
-    {
-      "filter_name": "SPECIFIC_VARIABLES_NULL_VALUES",
-      "policy": "CONTINUE",
-      "config": {
-        "variables": [
-          "Counter"
-        ]
-      }
+  "input_filters": {
+    "EMPTY_DATA": {"POLICY": "STOP"},
+    "SPECIFIC_VARIABLES_NULL_VALUES": {
+      "POLICY": "CONTINUE",
+      "config": {"variables": ["Counter"]}
    }
-  ],
-  "mlflow_transform_filters": [
-    {
-      "filter_name": "API_ERROR", # Required filter
-      "policy": "REPEAT"
-    },
-    {
-      "filter_name": "NAN_VALUES",
-      "policy": "STOP"
-    }
-  ],
-  "mlflow_predict_filters": [
-    {
-      "filter_name": "API_ERROR", # Required filter
-      "policy": "CONTINUE"
-    }
-  ],
-  "path_priority": [ # In case of multiple filters catch problems, this will determine the path to take
-    "STOP",
-    "CONTINUE",
-    "REPEAT"
-  ],
+  },
+  "mlflow_transform_filters": {
+    "API_ERROR": {"POLICY": "REPEAT"},
+    "NAN_VALUES": {"POLICY": "STOP"}
+  },
+  "mlflow_predict_filters": {
+    "API_ERROR": {"POLICY": "CONTINUE"}
+  },
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
  "active": true,
-  "datetime_columns": [ # Columns in data comming from query that are datetime
-    "timestamp",
-    "created_at"
-  ],
  "updated_at": {
-    "$date": "2025-08-27T18:35:01.600Z"
-  }
+    "$date": "2025-09-16T10:00:00.000Z"
+  },
+  "datetime_columns": ["timestamp", "created_at"],
+  "predictions_storage_policy": "lts:1"
+}
+```
+
+This is the configuration created by the Orchestrator in Temporal.
+
+```json
+{
+  "datetime_columns":["timestamp","created_at"],
+  "frequency":"15m",
+  "input_filters":{"EMPTY_DATA":{"config":{},"policy":"STOP"}},
+  "max_retry_policy":1,
+  "mlflow_predict_filters":{"API_ERROR":{"config":{},"policy":"CONTINUE"}},
+  "mlflow_transform_filters":{
+    "API_ERROR":{"config":{},"policy":"CONTINUE"},
+    "EMPTY_DATA":{"config":{},"policy":"STOP"}
+  },
+  "model_config":{
+    "is_compressed":true,
+    "predict_flavor":"pyfunc",
+    "retention_minutes":60,
+    "retention_target":"artifact",
+    "transform_function_keyword":"transform"
+  },
+  "model_id":"352",
+  "model_name":"courier",
+  "opc_output_config":{},
+  "path_priority":["STOP","CONTINUE","REPEAT"],
+  "predictions_storage_policy":"lts:1",
+  "query":"select * from sientia_data.laborious_data where model_id = 352 order by \"timestamp\" desc limit 300;",
+  "retention_time":3600,
+  "schedule_name":"laborious-courier",
+  "schema":"sientia_data",
+  "table_name":"predictions",
+  "updated_at":"2025-09-12 19:35:01.600000+0000",
+  "workflow_type":"predictions_batch"
 }
 ```

@@ -667,7 +717,7 @@ laborious/
 │   ├── gates.py            # Data quality gates and filtering
 │   ├── mlflow.py           # MLFlow model operations
 │   └── opc.py              # OPC server operations
-├── workflow/                # Temporal workflow definitions
+├── workflows/               # Temporal workflow definitions
 │   ├── predictions_batch.py # Main batch prediction workflow
 │   ├── minimal_retrain.py  # Model retraining workflow
 │   └── sub_workflows/      # Sub-workflow implementations
--- a/laborious/activities/gates.py
+++ b/laborious/activities/gates.py
@@ -8,6 +8,7 @@ with workflow.unsafe.imports_passed_through():
    from sientia_do.temporal.activities.base import BaseActivity
    from sientia_do.observability.logger import Logger
    from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now
+    from sientia_do.formatters import create_sample_dict
    from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter
    from typing import Any
    from laborious.utils.filters.conditional_filters import (
@@ -41,6 +42,7 @@ mlflow_response_filter_functions = {
 # MLFlow content filter function mappings
 mlflow_content_filter_functions = {
    'NAN_VALUES': nan_values_filter,
+    'EMPTY_DATA': filter_empty_data,
    'path_confidence': {
        'STOP': -1,
        'CONTINUE': 18,
@@ -121,15 +123,13 @@ class Gates(BaseActivity):

        self.info("Performing input gate...", metadata)

-        self.debug(f"Input data: {input_data}", metadata)
-
        filters = input_data['filters']
        data = DataFrame(input_data['data'])
        path_priority = input_data['path_priority']

        filter_output = []

-        self.debug(f"Input data:\n {data}", metadata)
+        self.debug(f"Input data: {data.head(5).to_string()}", metadata)
        self.debug(f"Filters: {filters}", metadata)

        # Apply each configured filter
@@ -206,7 +206,8 @@ class Gates(BaseActivity):

        filter_output = []

-        self.debug(f"Input data:\n {data}", metadata)
+        self.debug(
+            f"Input data: \n {create_sample_dict(data, max_items=5, max_depth=2)}", metadata)
        self.debug(f"Filters: {filters}", metadata)

        comments = []
@@ -291,8 +292,8 @@ class Gates(BaseActivity):

        filter_output = []

-        self.debug(f"Input data:\n {data}", metadata)
-        self.debug(f"Filters: {filters}", metadata)
+        self.debug(f"Input data:\n {data.head(5).to_string()}", metadata)
+        self.debug(f"Filters: \n {create_sample_dict(filters)}", metadata)

        for fil, config in filters.items():
            if fil not in mlflow_content_filter_functions:
@@ -403,36 +404,36 @@ class Gates(BaseActivity):

        data = DataFrame(input_data['data'])

+        # Create timestamp column from index and reset index
+        data['timestamp'] = data.index
+        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)

        policy_type, policy_value = self.get_prediction_store_policy(
            prediction_store_policy, metadata)

        # If data has no timestamp, we use the default timestamp and not sort the data
-        if 'timestamp' not in data.columns:
-            self.warning(
-                "Data has no timestamp, using default timestamp", metadata)
-            data['timestamp'] = input_data['timestamp']
-        else:
-            self.debug(
-                f"Data has timestamp, sorting data by timestamp", metadata)
+        self.info(
+            f"Sorting data by timestamp and applying policy: {policy_type}:{policy_value}", metadata)

-            # If policy_type is lts, we need to sort the data by timestamp descending and take the first policy_value rows
-            if policy_type == 'lts':
-                self.debug(
-                    f"Sorting data by timestamp descending", metadata)
-                data = data.sort_values(by='timestamp', ascending=False)
-            # If policy_type is erl, we need to sort the data by timestamp ascending and take the first policy_value rows
-            elif policy_type == 'erl':
-                self.debug(
-                    f"Sorting data by timestamp ascending", metadata)
-                data = data.sort_values(by='timestamp', ascending=True)
-            else:
-                self.error(
-                    f"Invalid policy type: {policy_type}, using default policy", metadata)
-                raise ValueError(
-                    f"Invalid policy type: {policy_type}")
+        # If policy_type is lts, we need to sort the data by timestamp descending and take the first policy_value rows
+        if policy_type == 'lts':
+            self.debug(
+                "Sorting data by timestamp descending", metadata)
+            data = data.sort_values(by='timestamp', ascending=False)
+        # If policy_type is erl, we need to sort the data by timestamp ascending and take the first policy_value rows
+        elif policy_type == 'erl':
+            self.debug(
+                "Sorting data by timestamp ascending", metadata)
+            data = data.sort_values(by='timestamp', ascending=True)
+        else:
+            self.error(
+                f"Invalid policy type: {policy_type}, using default policy", metadata)
+            raise ValueError(
+                f"Invalid policy type: {policy_type}")

        data = data.head(int(policy_value))

@@ -443,8 +444,8 @@ class Gates(BaseActivity):
        data = data.sort_values(by='timestamp', ascending=False)
        data = data.reset_index(drop=True)

-        self.info(f"Prediction formatted: {data.size} rows", metadata)
-        self.debug(f"Prediction data: {data.to_string()}", metadata)
+        self.info(f"Prediction formatted: {len(data)} rows", metadata)
+        self.debug(f"Prediction data: {data.head(5).to_string()}", metadata)

        return data.to_dict()

@@ -520,7 +521,7 @@ class Gates(BaseActivity):

        data = DataFrame(input_data['data'])

-        self.debug(f"Input data: {data.to_string()}", metadata)
+        self.debug(f"Input data: {data.head(5).to_string()}", metadata)

        if data.empty:
            return now().strftime(DATETIME_FORMAT_WITH_TZ)
--- a/laborious/activities/mlflow.py
+++ b/laborious/activities/mlflow.py
@@ -1,12 +1,15 @@
-import json
 from temporalio import activity, workflow


 with workflow.unsafe.imports_passed_through():
+    from datetime import datetime
+    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
    from sientia_do.observability.logger import Logger
+    from sientia_do.formatters import create_sample_dict
    from laborious.utils.repository.model_repository import MLFlowRepository
    from typing import Any
    import numpy as np
@@ -57,7 +60,7 @@ class MLFlow(BaseActivity):
        self.mlflow_password = mlflow_password

        self.model_monitoring_repository = MLFlowRepository(
-            f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password, logger.base_logger
+            f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password, logger
        )

    @activity.defn(name="request_transform")
@@ -94,10 +97,10 @@ class MLFlow(BaseActivity):
        self.info('Transforming data...', metadata)
        data = DataFrame(input_data['data'])
        model_name = input_data['model_name']
-        model_retention = input_data['model_retention']
+        model_config = input_data.get('model_config', {})

        self.debug("Raw input data:", metadata)
-        self.debug(data, metadata)
+        self.debug(data.head(5).to_string(), metadata)

        # Sort by created_at in descending order and keep first occurrence of each variable/timestamp pair
        data = data.sort_values('created_at', ascending=False).drop_duplicates(
@@ -113,14 +116,18 @@ class MLFlow(BaseActivity):
        data.columns.name = None

        self.debug("Processed input data:", metadata)
-        self.debug(data, metadata)
+        self.debug(data.head(5).to_string(), metadata)

        # Request transformation from MLFlow model
        response_data = self.model_monitoring_repository.transform(
-            model_name, data, model_retention)
+            model_name, data, model_config, metadata
+        )

-        self.debug("Transform response data:", metadata)
-        self.debug(json.dumps(response_data, indent=4), metadata)
+        self.debug(
+            f"Transform raw response data: \n {create_sample_dict(response_data, max_items=5, max_depth=5)}", metadata)
+
+        self.debug(
+            f"Transform response data: \n {create_sample_dict(response_data, max_items=5, max_depth=5)}", metadata)

        self.info("Data transformed successfully", metadata)

@@ -160,19 +167,24 @@ class MLFlow(BaseActivity):
        self.info('Predicting data...', metadata)
        data = DataFrame(input_data['data'])
        model_name = input_data['model_name']
-        model_retention = input_data['model_retention']
+        model_config = input_data.get('model_config', {})

-        self.debug(data, metadata)
+        self.debug(f"Input data for: \n {data.head(5).to_string()}", metadata)

        # Convert numpy.nan to None for model compatibility
        data.replace(np.nan, None, inplace=True)

+        data['timestamp'] = data.index
+        data['timestamp'] = to_datetime(
+            data['timestamp'], format=DATETIME_FORMAT_WITH_TZ).dt.strftime(DATETIME_FORMAT)
+
        # Request prediction from MLFlow model
        response_data = self.model_monitoring_repository.predict(
-            model_name, data, model_retention)
+            model_name, data, model_config, metadata
+        )

-        self.debug("Prediction response data:", metadata)
-        self.debug(json.dumps(response_data, indent=4), metadata)
+        self.debug(
+            f"Prediction response data: \n {create_sample_dict(response_data, max_items=5, max_depth=5)}", metadata)

        self.info("Data predicted successfully", metadata)

--- a/laborious/utils/connectors_config.py
+++ b/laborious/utils/connectors_config.py
@@ -1,205 +1,99 @@
-"""
-Connectors Configuration Module
-
-This module provides configuration management for all external service connectors
-used by the Sientia DataOps Laborious system. It centralizes configuration
-for databases, MLFlow servers, OPC servers, and other external dependencies.
-
-The module implements configuration builders for:
-1. PostgreSQL database connections
-2. MLFlow model serving endpoints
-3. OPC server configurations
-4. MongoDB notification systems
-
-Key Features:
- Environment variable-based configuration
- Default value management for development
- Connection pool configuration
- Security credential management
- Configuration validation and error handling
- Support for multiple service instances
-
-Configuration Sources:
- Environment variables for production deployment
- Default values for local development
- Kubernetes secrets integration
- Configurable connection parameters
-
-Environment Variables:
- POSTGRES_*: PostgreSQL connection parameters
- MLFLOW_*: MLFlow server parameters
- OPC_*: OPC server configuration
- MONGODB_*: MongoDB connection parameters
-
-Dependencies:
- os: Environment variable access
- typing: Type hints and annotations
-"""
-
-import os
+from os import getenv
+import json
 from typing import Dict, Any


 def build_postgres_config() -> Dict[str, Any]:
    """
    Build PostgreSQL database configuration from environment variables.
-    
+
    This function constructs a PostgreSQL configuration dictionary from
    environment variables with sensible defaults for local development.
    It handles connection pool configuration and security parameters.
-    
+
    Environment Variables:
        POSTGRES_HOST: Database hostname (default: localhost)
        POSTGRES_PORT: Database port (default: 5432)
        POSTGRES_USER: Database username (default: sientia)
        POSTGRES_PASSWORD: Database password (default: sientia)
        POSTGRES_DBNAME: Database name (default: sientia)
-        POSTGRES_MIN_CONNECTIONS: Minimum connection pool size (default: 1)
-        POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 10)
-    
+        POSTGRES_MIN_CONNECTIONS: Minimum connection pool size (default: 5)
+        POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 20)
+
    Returns:
        dict: PostgreSQL configuration dictionary with all required parameters
-        
-    Example:
-        >>> config = build_postgres_config()
-        >>> print(config)
-        {
-            'host': 'localhost',
-            'port': 5432,
-            'user': 'sientia',
-            'password': 'sientia',
-            'dbname': 'sientia',
-            'min_connections': 1,
-            'max_connections': 10
-        }
-    
-    Note:
-        In production, ensure all required environment variables are set
-        with appropriate values for your database environment.
    """
    return {
-        'host': os.getenv('POSTGRES_HOST', 'localhost'),
-        'port': int(os.getenv('POSTGRES_PORT', '5432')),
-        'user': os.getenv('POSTGRES_USER', 'sientia'),
-        'password': os.getenv('POSTGRES_PASSWORD', 'sientia'),
-        'dbname': os.getenv('POSTGRES_DBNAME', 'sientia'),
-        'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '1')),
-        'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '10'))
+        'host': getenv('POSTGRES_HOST', 'localhost'),
+        'port': int(getenv('POSTGRES_PORT', '5432')),
+        'user': getenv('POSTGRES_USER', 'sientia'),
+        'password': getenv('POSTGRES_PASSWORD', 'sientia'),
+        'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
+        'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
+        'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20'))
    }


 def build_mlflow_config() -> Dict[str, Any]:
    """
    Build MLFlow server configuration from environment variables.
-    
+
    This function constructs an MLFlow configuration dictionary from
    environment variables with sensible defaults for local development.
    It handles server connection and authentication parameters.
-    
+
    Environment Variables:
-        MLFLOW_HOST: MLFlow server hostname (default: localhost)
-        MLFLOW_PORT: MLFlow server port (default: 5000)
-        MLFLOW_USERNAME: MLFlow username (default: admin)
-        MLFLOW_PASSWORD: MLFlow password (default: admin)
-    
+        MLFLOW_HOST: MLFlow server hostname (default: http://localhost)
+        MLFLOW_PORT: MLFlow server port (default: 5080)
+        MLFLOW_USERNAME: MLFlow username (default: aignosi)
+        MLFLOW_PASSWORD: MLFlow password (default: aignosi)
+
    Returns:
        dict: MLFlow configuration dictionary with all required parameters
-        
-    Example:
-        >>> config = build_mlflow_config()
-        >>> print(config)
-        {
-            'host': 'localhost',
-            'port': 5000,
-            'username': 'admin',
-            'password': 'admin'
-        }
-    
-    Note:
-        In production, ensure all required environment variables are set
-        with appropriate values for your MLFlow server environment.
-        Consider using secure authentication methods for production deployments.
    """
    return {
-        'host': os.getenv('MLFLOW_HOST', 'localhost'),
-        'port': int(os.getenv('MLFLOW_PORT', '5000')),
-        'username': os.getenv('MLFLOW_USERNAME', 'admin'),
-        'password': os.getenv('MLFLOW_PASSWORD', 'admin')
+        'host': getenv('MLFLOW_HOST', 'http://localhost'),
+        'port': int(getenv('MLFLOW_PORT', '5080')),
+        'username': getenv('MLFLOW_USERNAME', 'aignosi'),
+        'password': getenv('MLFLOW_PASSWORD', 'aignosi')
    }


 def build_opc_config() -> Dict[str, Any]:
    """
    Build OPC server configuration from environment variables.
-    
+
    This function constructs an OPC server configuration dictionary from
    environment variables. It supports both single server and multi-server
    configurations with flexible parameter handling.
-    
+
    Environment Variables:
        OPC_CONFIG: JSON string containing multiple OPC server configurations
-        OPC_URL: Single OPC server URL (fallback)
-        OPC_NAME: Single OPC server name (fallback)
-        OPC_SERVER_URI: Single OPC server URI (fallback)
-        OPC_CERT_PATH: Client certificate path (fallback)
-        OPC_PRIVATE_KEY_PATH: Client private key path (fallback)
-        OPC_SERVER_CERT_PATH: Server certificate path (fallback)
-        OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback)
-    
+        OPC_ID: OPC server ID (fallback, default: 1)
+        OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840)
+        OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840)
+        OPC_CERT_PATH: Client certificate path (fallback, default: None)
+        OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None)
+        OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None)
+        OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback, default: 120)
+
    Returns:
        dict: OPC server configuration dictionary
-        
-    Configuration Modes:
-        1. Multi-server: Use OPC_CONFIG environment variable with JSON string
-        2. Single server: Use individual OPC_* environment variables
-        
-    Example Multi-server Configuration:
-        >>> # Set OPC_CONFIG environment variable
-        >>> os.environ['OPC_CONFIG'] = '''
-        ... {
-        ...     "opc_server_1": {
-        ...         "url": "opc.tcp://server1:4840",
-        ...         "name": "Server1",
-        ...         "server_uri": "urn:server1:opcua",
-        ...         "cert_path": "/path/to/cert.pem",
-        ...         "private_key_path": "/path/to/key.pem",
-        ...         "server_cert_path": "/path/to/server_cert.pem",
-        ...         "reconnection_interval": 5000
-        ...     }
-        ... }
-        ... '''
-        >>> config = build_opc_config()
-        
-    Example Single Server Configuration:
-        >>> # Set individual environment variables
-        >>> os.environ['OPC_URL'] = 'opc.tcp://localhost:4840'
-        >>> os.environ['OPC_NAME'] = 'LocalServer'
-        >>> config = build_opc_config()
-    
-    Note:
-        For production deployments, prefer the OPC_CONFIG approach for
-        multiple servers and ensure all certificate paths are properly configured.
    """
-    # Check for multi-server configuration
-    opc_config = os.getenv('OPC_CONFIG')
-    if opc_config:
-        try:
-            import json
-            return json.loads(opc_config)
-        except (json.JSONDecodeError, ImportError) as e:
-            # Fall back to single server configuration if JSON parsing fails
-            pass
-    
-    # Single server configuration fallback
+    opc_raw = getenv('OPC_CONFIG', None)
+
+    if opc_raw:
+        return json.loads(opc_raw)
+
    return {
-        'default': {
-            'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'),
-            'name': os.getenv('OPC_NAME', 'DefaultServer'),
-            'server_uri': os.getenv('OPC_SERVER_URI', 'urn:default:opcua'),
-            'cert_path': os.getenv('OPC_CERT_PATH', ''),
-            'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', ''),
-            'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', ''),
-            'reconnection_interval': int(os.getenv('OPC_RECONNECTION_INTERVAL', '5000'))
+        getenv('OPC_ID', '1'): {
+            'id': getenv('OPC_ID', '1'),
+            'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'),
+            'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
+            'cert_path': getenv('OPC_CERT_PATH', None),
+            'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None),
+            'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None),
+            'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120'))
        }
    }

@@ -207,32 +101,29 @@ def build_opc_config() -> Dict[str, Any]:
 def build_mongodb_config() -> Dict[str, Any]:
    """
    Build MongoDB configuration from environment variables.
-    
+
    This function constructs a MongoDB configuration dictionary from
    environment variables with sensible defaults for local development.
    It handles connection string and database name configuration.
-    
+
    Environment Variables:
-        MONGODB_URL: MongoDB connection URI (default: localhost:27017)
-        MONGODB_DATABASE: MongoDB database name (default: sientia)
-    
+        MONGODB_USERNAME: MongoDB username (default: root)
+        MONGODB_PASSWORD: MongoDB password (default: wKZDbMNU1c)
+        MONGODB_URL: MongoDB connection URI (default: localhost:27018)
+        MONGODB_DATABASE_NAME: MongoDB database name (default: sientia)
+        MONGODB_TTL_INDEX_HOURS: TTL index duration in hours (default: 1)
+
    Returns:
        dict: MongoDB configuration dictionary with connection parameters
-        
-    Example:
-        >>> config = build_mongodb_config()
-        >>> print(config)
-        {
-            'connection_string': 'localhost:27017',
-            'database_name': 'sientia'
-        }
-    
-    Note:
-        In production, ensure the MONGODB_URL environment variable is set
-        with a proper MongoDB connection string including authentication
-        if required by your MongoDB deployment.
    """
+    username = getenv('MONGODB_USERNAME', 'root')
+    password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
+    uri = getenv('MONGODB_URL', 'localhost:27018')
+
+    connection_string = f'mongodb://{username}:{password}@{uri}'
+
    return {
-        'connection_string': os.getenv('MONGODB_URL', 'localhost:27017'),
-        'database_name': os.getenv('MONGODB_DATABASE', 'sientia')
+        'connection_string': connection_string,
+        'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
+        'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600
    }
--- a/laborious/utils/filters/conditional_filters.py
+++ b/laborious/utils/filters/conditional_filters.py
@@ -1,197 +1,45 @@
-"""
-Conditional Data Filters Module
-
-This module provides conditional data filtering functions for the Sientia DataOps Laborious system.
-It implements data quality validation filters that can be applied to input data before
-ML operations to ensure data integrity and quality.
-
-The module implements filters for:
-1. Empty data detection and validation
-2. Specific variable null value checking
-3. Configurable data quality rules
-4. Flexible filter configuration
-
-Key Features:
- Configurable filter policies and thresholds
- Multiple data quality validation rules
- Flexible configuration options
- Comprehensive error handling
- Performance-optimized filtering
-
-Filter Types:
- EMPTY_DATA: Detects empty or insufficient data sets
- SPECIFIC_VARIABLES_NULL_VALUES: Validates specific variable null values
- Custom filters can be added for specific validation needs
-
-Dependencies:
- pandas.DataFrame: Data manipulation and processing
- typing: Type hints and annotations
-"""
-
-from typing import Any, Dict, List
 from pandas import DataFrame


-def filter_empty_data(data: DataFrame, config: Dict[str, Any]) -> bool:
+def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool:
    """
-    Filter data based on empty data conditions.
-    
-    This function checks if the input data meets minimum requirements for
-    processing. It can validate data size, completeness, and other quality
-    metrics to ensure sufficient data is available for ML operations.
-    
-    The filter implements multiple validation criteria:
-    1. Data frame size validation
-    2. Row count validation
-    3. Column completeness validation
-    4. Configurable threshold checking
-    
+    Filter to check if specific variables contain null values.
+
+    This function examines a DataFrame to determine if any of the specified variables
+    contain null (NaN) values. It returns True if null values are found for any of
+    the specified variables, False otherwise.
+
    Args:
-        data: Input data as pandas DataFrame
-        config: Filter configuration dictionary
-            Required keys:
-                - min_rows (int, optional): Minimum number of rows required
-                - min_columns (int, optional): Minimum number of columns required
-                - min_data_points (int, optional): Minimum total data points required
-    
+        data (DataFrame): The pandas DataFrame to be examined. Must contain columns
+            named 'variable' and 'value'.
+        config (dict): Configuration dictionary containing the following key:
+            - variables (list): List of variable names to check for null values
+
    Returns:
-        bool: True if data should be filtered (fails quality check), False otherwise
-        
-    Filter Logic:
-        - Returns True (filter) if data is empty or below thresholds
-        - Returns False (pass) if data meets quality requirements
-        - Handles missing configuration gracefully with defaults
-        
-    Example:
-        >>> import pandas as pd
-        >>> df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})
-        >>> config = {'min_rows': 2, 'min_columns': 2}
-        >>> result = filter_empty_data(df, config)
-        >>> print(result)
-        False  # Data passes filter
-        
-        >>> empty_df = pd.DataFrame()
-        >>> result = filter_empty_data(empty_df, config)
-        >>> print(result)
-        True   # Data fails filter
-        
-    Default Thresholds:
-        - min_rows: 1 (at least one row required)
-        - min_columns: 1 (at least one column required)
-        - min_data_points: 1 (at least one data point required)
+        bool: True if any of the specified variables contain null values,
+              False if none of the specified variables contain null values.
+
    """
-    # Check if data is completely empty
-    if data.empty:
-        return True
-    
-    # Get configuration with defaults
-    min_rows = config.get('min_rows', 1)
-    min_columns = config.get('min_columns', 1)
-    min_data_points = config.get('min_data_points', 1)
-    
-    # Check row count
-    if len(data) < min_rows:
-        return True
-    
-    # Check column count
-    if len(data.columns) < min_columns:
-        return True
-    
-    # Check total data points
-    if data.size < min_data_points:
-        return True
-    
-    # Data passes all quality checks
-    return False
+    return not data[
+        data['variable'].isin(config['variables']) & data['value'].isna()].empty


-def filter_specific_variables_null_values(data: DataFrame, config: Dict[str, Any]) -> bool:
+def filter_empty_data(data: DataFrame, _config: dict) -> bool:
    """
-    Filter data based on null values in specific variables.
-    
-    This function checks for null values in specified variables and determines
-    if the data quality is sufficient for processing. It can validate
-    individual columns or groups of columns for data completeness.
-    
-    The filter implements variable-specific validation:
-    1. Individual variable null value checking
-    2. Configurable null value thresholds
-    3. Multiple variable validation
-    4. Flexible threshold configuration
-    
+    Filter to check if the DataFrame is empty.
+
+    This function determines whether the provided DataFrame contains any data.
+    It's a simple utility function that can be used in conditional logic to
+    handle cases where no data is available.
+
    Args:
-        data: Input data as pandas DataFrame
-        config: Filter configuration dictionary
-            Required keys:
-                - variables (list): List of variable names to check
-                - max_null_ratio (float, optional): Maximum allowed null value ratio (0.0 to 1.0)
-                - max_null_count (int, optional): Maximum allowed null value count
-    
+        data (DataFrame): The pandas DataFrame to be checked for emptiness.
+        _config (dict): Configuration dictionary (unused in this function).
+            The underscore prefix indicates this parameter is required for
+            interface consistency but not used in the implementation.
+
    Returns:
-        bool: True if data should be filtered (fails quality check), False otherwise
-        
-    Filter Logic:
-        - Returns True (filter) if null value thresholds are exceeded
-        - Returns False (pass) if null values are within acceptable limits
-        - Handles missing variables gracefully
-        - Supports both ratio and count-based thresholds
-        
-    Example:
-        >>> import pandas as pd
-        >>> df = pd.DataFrame({
-        ...     'temperature': [25.5, None, 27.0, 26.5],
-        ...     'humidity': [60.0, 65.0, None, 62.0]
-        ... })
-        >>> config = {
-        ...     'variables': ['temperature', 'humidity'],
-        ...     'max_null_ratio': 0.25
-        ... }
-        >>> result = filter_specific_variables_null_values(df, config)
-        >>> print(result)
-        False  # Data passes filter (null ratio = 0.25, which equals max)
-        
-        >>> config = {
-        ...     'variables': ['temperature', 'humidity'],
-        ...     'max_null_ratio': 0.20
-        ... }
-        >>> result = filter_specific_variables_null_values(df, config)
-        >>> print(result)
-        True   # Data fails filter (null ratio = 0.25, exceeds max of 0.20)
-        
-    Default Thresholds:
-        - max_null_ratio: 0.5 (50% null values allowed)
-        - max_null_count: None (no count-based limit by default)
-        
-    Note:
-        If both max_null_ratio and max_null_count are specified, the filter
-        will trigger if either threshold is exceeded.
+        bool: True if the DataFrame is empty (has no rows), False if it contains data.
+
    """
-    # Get configuration
-    variables = config.get('variables', [])
-    max_null_ratio = config.get('max_null_ratio', 0.5)
-    max_null_count = config.get('max_null_count', None)
-    
-    # Check if variables exist in data
-    if not variables:
-        return False  # No variables specified, pass filter
-    
-    # Validate each specified variable
-    for variable in variables:
-        if variable not in data.columns:
-            continue  # Skip variables that don't exist in data
-        
-        # Calculate null value statistics
-        null_count = data[variable].isnull().sum()
-        total_count = len(data[variable])
-        null_ratio = null_count / total_count if total_count > 0 else 0.0
-        
-        # Check ratio threshold
-        if null_ratio > max_null_ratio:
-            return True
-        
-        # Check count threshold (if specified)
-        if max_null_count is not None and null_count > max_null_count:
-            return True
-    
-    # All variables pass null value checks
-    return False
+    return data.empty
--- a/laborious/utils/repository/model_repository.py
+++ b/laborious/utils/repository/model_repository.py
@@ -16,16 +16,57 @@ import pandas as pd
 import mlflow
 from os import makedirs, path, remove
 from sientia.ModelServing import ModelServing
+from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
+from sientia_do.observability.logger import Logger


 class MLFlowRepository():
-    def __init__(self, host, username, password, logger):
+    def __init__(self, host, username, password, logger: Logger):

        self.model_serving = ModelServing(tracking_uri=host,
                                          username=username, password=password,
                                          logger=logger)
+        self.logger = logger

-    def transform(self, model_name: str, data: pd.DataFrame, model_retention: int):
+    def detect_and_parse_datetime_index(self, data: pd.DataFrame, metadata: dict) -> pd.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.logger.custom_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:
+                pd.to_datetime(data.index, format=DATETIME_FORMAT_WITH_TZ)
+            except ValueError:
+                raise ValueError(
+                    f"{message}")
+
+        elif index_type == datetime or index_type == pd.Timestamp:
+            data.index = data.index.strftime(DATETIME_FORMAT_WITH_TZ)
+        else:
+            raise ValueError(
+                f"{message}")
+
+        return data
+
+    def transform(self, model_name: str, data: pd.DataFrame, model_config: dict, metadata: dict) -> dict:
        """
        Transform data using a model.

@@ -39,10 +80,30 @@ class MLFlowRepository():
        """

        try:
+            self.logger.custom_debug(
+                f"Data received for model transformation: {data.to_csv()}", metadata)
+
+            model_retention = model_config.get('retention_minutes', 0)
+            flavor = model_config.get('transform_flavor', 'sklearn')
+            compressed = model_config.get('is_compressed', False)
+            retention_target = model_config.get('retention_target', 'model')
+            transform_keyword = model_config.get(
+                'transform_function_keyword', 'predict')
+
+            transformed_data = self.model_serving.get_cached_transform(
+                model_name, data, model_retention, flavor,
+                compressed, retention_target, transform_keyword
+            )
+
+            self.logger.custom_debug(
+                f"Data received from model transformation: {transformed_data.to_csv()}", metadata)
+
+            transformed_data = self.detect_and_parse_datetime_index(
+                transformed_data, metadata)
+
            return {
                'success': True,
-                'content': self.model_serving.get_cached_transform(
-                    model_name, data, model_retention).to_dict()
+                'content': transformed_data.to_dict()
            }

        except Exception as e:
@@ -54,7 +115,7 @@ class MLFlowRepository():
                }
            }

-    def predict(self, model_name: str, data: pd.DataFrame, model_retention: int):
+    def predict(self, model_name: str, data: pd.DataFrame, model_config: dict, metadata: dict) -> dict:
        """
        Predict data using a model.

@@ -67,12 +128,26 @@ class MLFlowRepository():
            - dict: A dictionary containing the predicted data.
        """
        try:
+            model_retention = model_config.get('retention_minutes', 0)
+            flavor = model_config.get('predict_flavor', 'pyfunc')
+            compressed = model_config.get('is_compressed', False)
+            retention_target = model_config.get('retention_target', 'model')
+
+            input_index = data.index
            start_time = datetime.now()
+
+            self.logger.custom_debug(
+                f"Data received for model prediction: {data.to_csv()}", metadata)
            data = self.model_serving.get_cached_predict(
-                model_name, data, model_retention)
+                model_name, data, model_retention, flavor,
+                compressed, retention_target
+            )

            end_time = datetime.now()
            data = pd.DataFrame(data, columns=['prediction'])
+            self.logger.custom_debug(
+                f"Data received from model prediction: {data.to_csv()}", metadata)
+            data.index = input_index
            data['response_time'] = (end_time - start_time).total_seconds()

            return {
--- a/laborious/workflows/predictions_batch.py
+++ b/laborious/workflows/predictions_batch.py
@@ -29,22 +29,6 @@ class PredictionsBatch():
        2. Configuration Preparation: Sets up prediction parameters and filters
        3. Workflow Delegation: Spawns PredictionProcess child workflow
        4. Error Handling: Implements comprehensive error handling and retry policies
-
-    Example:
-        >>> # Start the workflow
-        >>> await client.start_workflow(
-        ...     PredictionsBatch.run,
-        ...     id="batch_pred_001",
-        ...     task_queue="predictions_batch-queue",
-        ...     input_data={
-        ...         "schedule_name": "hourly_predictions",
-        ...         "model_name": "temperature_model",
-        ...         "model_id": "temp_001",
-        ...         "query": "SELECT * FROM sensor_data WHERE timestamp > NOW() - INTERVAL '1 hour'",
-        ...         "schema": {"timestamp": "datetime", "temperature": "float"},
-        ...         "table_name": "predictions"
-        ...     }
-        ... )
    """

    @workflow.run
@@ -103,7 +87,7 @@ class PredictionsBatch():
                'datetime_columns': input_data.get('datetime_columns', [])
            },
            retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
+            start_to_close_timeout=timedelta(seconds=300)
        )

        # Prepare input for prediction_process workflow
@@ -129,9 +113,11 @@ class PredictionsBatch():
                    'POLICY': 'STOP'
                }
            }),
-            'model_retention': input_data.get('model_retention', 60),
+            'model_config': input_data.get('model_config', {}),
            'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']),
-            'opc_output_config': input_data.get('opc_output_config', {})
+            'opc_output_config': input_data.get('opc_output_config', {}),
+            'prediction_store_policy': input_data.get(
+                'prediction_store_policy', 'lts:1')
        }

        # Execute prediction process workflow
--- a/laborious/workflows/sub_workflows/format_and_export_prediction.py
+++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py
@@ -79,8 +79,7 @@ class FormatAndExportPrediction():
                    'timestamp': input_data['timestamp'],
                    'model_id': input_data['model_id'],
                    'prediction_confidence': prediction_confidence,
-                    'prediction_store_policy': input_data.get(
-                        'prediction_store_policy', 'lts:1')
+                    'prediction_store_policy': input_data['prediction_store_policy']
                },
                retry_policy=retry_policy,
                start_to_close_timeout=timedelta(seconds=60)
--- a/laborious/workflows/sub_workflows/prediction_process.py
+++ b/laborious/workflows/sub_workflows/prediction_process.py
@@ -79,7 +79,7 @@ class PredictionProcess():
        data = input_data['data']
        model_id = input_data['model_id']
        model_name = input_data['model_name']
-        model_retention = input_data['model_retention']
+        model_config = input_data.get('model_config', {})

        # Get last timestamp for incremental processing
        last_timestamp = await workflow.execute_local_activity_method(
@@ -120,10 +120,10 @@ class PredictionProcess():
                **metadata,
                'data': data,
                'model_name': model_name,
-                'model_retention': model_retention
+                'model_config': model_config
            },
            retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(minutes=1),
+            start_to_close_timeout=timedelta(minutes=5),
        )

        # Validate MLFlow transform response
@@ -172,10 +172,10 @@ class PredictionProcess():
                **metadata,
                'data': transformed_data,
                'model_name': model_name,
-                'model_retention': model_retention
+                'model_config': model_config
            },
            retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(minutes=1),
+            start_to_close_timeout=timedelta(minutes=5),
        )

        # Validate MLFlow prediction response
@@ -209,11 +209,12 @@ class PredictionProcess():
                'timestamp': last_timestamp,
                'model_id': model_id,
                'model_name': model_name,
-                'model_retention': model_retention,
+                'model_config': model_config,
                'opc_output_config': input_data['opc_output_config'],
                'schema': input_data['schema'],
                'table_name': input_data['table_name'],
-                'comment': comment
+                'comment': comment,
+                'prediction_store_policy': input_data['prediction_store_policy']
            }
        )

@@ -248,7 +249,7 @@ class PredictionProcess():
        table_name = input_data['table_name']
        model_id = input_data['model_id']
        model_name = input_data['model_name']
-        model_retention = input_data['model_retention']
+        model_config = input_data.get('model_config', {})

        path_flag = path_flag.upper() if path_flag else ''

@@ -282,11 +283,12 @@ class PredictionProcess():
                    'timestamp': last_timestamp,
                    'model_id': model_id,
                    'model_name': model_name,
-                    'model_retention': model_retention,
+                    'model_config': model_config,
                    'schema': schema,
                    'table_name': table_name,
                    'comment': comment,
-                    'opc_output_config': input_data['opc_output_config']
+                    'opc_output_config': input_data['opc_output_config'],
+                    'prediction_store_policy': input_data['prediction_store_policy']
                }
            )
            return True
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,6 +3,6 @@ psycopg2-binary
 sqlalchemy
 asyncua
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.4
-git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.12
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6
+git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0
 prometheus-client
--- a/requirements_prepared.txt
+++ b/requirements_prepared.txt
@@ -1,8 +0,0 @@
-temporalio
-psycopg2-binary
-sqlalchemy
-asyncua
-redis
-git+https://github.com/Aignosi/sientia-dataops-library.git@1.4.4
-git+https://github.com/Aignosi/sientia-mlops-library.git@0.38.12
-prometheus-client
--- a/tests.ipynb
+++ b/tests.ipynb
@@ -2,7 +2,7 @@
 "cells": [
  {
   "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 1,
   "id": "b10e5c25",
   "metadata": {},
   "outputs": [
@@ -110,12 +110,12 @@
      "  0.06282208 -0.79004057 -0.20289499  0.83467535  0.26166448 -0.64498684\n",
      " -0.32228873 -0.61679398 -0.95035374  0.85492092 -0.10358534 -0.38492986\n",
      "  0.19695438 -0.98537109 -0.44395579  0.40606693  0.26753955]\n",
-      "    Counter   Rollout  CounterPlusRollout           Timestamp\n",
-      "0  0.000000  0.000000            0.000000 2025-01-01 00:00:00\n",
-      "1  0.547912  0.284967            0.832879 2025-01-01 00:00:01\n",
-      "2  0.425669  0.842959            1.268628 2025-01-01 00:00:02\n",
-      "3  1.142865  0.112064            1.254928 2025-01-01 00:00:03\n",
-      "4  1.537601  0.184200            1.721801 2025-01-01 00:00:04\n"
+      "     Counter    Rollout  CounterPlusRollout           Timestamp\n",
+      "0  37.270658 -59.474345          -22.203688 2025-01-01 00:00:00\n",
+      "1  43.561734 -54.184658          -10.622924 2025-01-01 00:00:01\n",
+      "2  42.158150 -43.826926           -1.668776 2025-01-01 00:00:02\n",
+      "3  50.392926 -57.394165           -7.001239 2025-01-01 00:00:03\n",
+      "4  54.925249 -56.055140           -1.129892 2025-01-01 00:00:04\n"
     ]
    }
   ],
@@ -165,33 +165,299 @@
  },
  {
   "cell_type": "code",
-   "execution_count": 2,
-   "id": "e7c8eeb1",
+   "execution_count": 1,
+   "id": "c61be7ab",
   "metadata": {},
   "outputs": [
    {
-     "ename": "TypeError",
-     "evalue": "MLFlowRepository.__init__() missing 1 required positional argument: 'logger'",
-     "output_type": "error",
-     "traceback": [
-      "\u001b[31m---------------------------------------------------------------------------\u001b[39m",
-      "\u001b[31mTypeError\u001b[39m                                 Traceback (most recent call last)",
-      "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[2]\u001b[39m\u001b[32m, line 3\u001b[39m\n\u001b[32m      1\u001b[39m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mlaborious\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mutils\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mrepository\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mmodel_repository\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m MLFlowRepository\n\u001b[32m----> \u001b[39m\u001b[32m3\u001b[39m mlflow_repository = \u001b[43mMLFlowRepository\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m      4\u001b[39m \u001b[43m    \u001b[49m\u001b[43mhost\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mhttp://localhost:5080/\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[32m      5\u001b[39m \u001b[43m    \u001b[49m\u001b[43musername\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43maignosi\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[32m      6\u001b[39m \u001b[43m    \u001b[49m\u001b[43mpassword\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43maignosi\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\n\u001b[32m      7\u001b[39m \u001b[43m)\u001b[49m\n\u001b[32m      9\u001b[39m mlflow_repository.get_experiment_by_run_id(\u001b[33m\"\u001b[39m\u001b[33m1\u001b[39m\u001b[33m\"\u001b[39m)\n",
-      "\u001b[31mTypeError\u001b[39m: MLFlowRepository.__init__() missing 1 required positional argument: 'logger'"
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "[ 0.5479121  -0.12224312  0.71719584  0.39473606 -0.8116453   0.9512447\n",
+      "  0.5222794   0.57212861 -0.74377273 -0.09922812 -0.25840395  0.85352998\n",
+      "  0.28773024  0.64552323 -0.1131716  -0.54552256  0.10916957 -0.87236549\n",
+      "  0.65526234  0.2633288   0.51617548 -0.29094806  0.94139605  0.78624224\n",
+      "  0.55676699 -0.61072258 -0.06655799 -0.91239247 -0.69142102  0.36609791\n",
+      "  0.48952431  0.93501946 -0.34834928 -0.25908059 -0.06088838 -0.62105728\n",
+      " -0.74015699 -0.04859015 -0.5461813   0.33962799 -0.12569616  0.66535639\n",
+      "  0.4005302  -0.37526672  0.6645196   0.60952871 -0.22504324 -0.42334379\n",
+      "  0.36499101]\n",
+      "[-0.72049503 -0.6001836  -0.98527546  0.57384876  0.32970171  0.41033076\n",
+      "  0.56145806 -0.08216845  0.13748239 -0.720406   -0.77093985  0.33680592\n",
+      " -0.05780759  0.13047221  0.52999771  0.26943664  0.1071588   0.11841432\n",
+      " -0.3920998  -0.93836433 -0.12656522 -0.57083065 -0.18294271  0.70680615\n",
+      " -0.53212103 -0.88339452 -0.43723222 -0.41281248  0.32383303  0.1140643\n",
+      "  0.56779642  0.32862708 -0.18722628  0.62804077 -0.66605416 -0.95457585\n",
+      " -0.81990428  0.4447187  -0.07624554 -0.67745644  0.00208955 -0.69537579\n",
+      "  0.39264075 -0.10768745 -0.23795755 -0.39697582  0.26056519 -0.27637478\n",
+      " -0.82470016]\n",
+      "      Counter     Rollout     Square           Timestamp\n",
+      "0 -100.000000  100.000000 -37.646304 2025-01-01 00:00:00\n",
+      "1  -79.942159   79.589040 -72.994086 2025-01-01 00:00:05\n",
+      "2  -84.417207   62.586392 -30.252436 2025-01-01 00:00:10\n",
+      "3  -58.162265   34.674447   7.555532 2025-01-01 00:00:15\n",
+      "4  -43.711857   50.931053  26.035410 2025-01-01 00:00:20\n"
     ]
    }
   ],
   "source": [
-    "from laborious.utils.repository.model_repository import MLFlowRepository\n",
+    "import pandas as pd\n",
+    "import numpy as np\n",
    "\n",
-    "mlflow_repository = MLFlowRepository(\n",
-    "    host=\"http://localhost:5080/\",\n",
-    "    username=\"aignosi\",\n",
-    "    password=\"aignosi\"\n",
-    ")\n",
+    "# Set random seed for reproducibility\n",
+    "rng = np.random.default_rng(42)\n",
    "\n",
-    "mlflow_repository.get_experiment_by_run_id(\"1\")"
+    "size = 50\n",
+    "\n",
+    "# Generate random walks starting at 0\n",
+    "counter = np.zeros(size)\n",
+    "rollout = np.zeros(size)\n",
+    "square = np.zeros(size)\n",
+    "\n",
+    "# Generate random steps between -1 and 1\n",
+    "counter_steps = rng.uniform(-1, 1, size-1)\n",
+    "rollout_steps = rng.uniform(-1, 1, size-1)\n",
+    "square_steps = rng.uniform(-1, 1, size-1)\n",
+    "\n",
+    "print(counter_steps)\n",
+    "print(rollout_steps)\n",
+    "\n",
+    "# Calculate cumulative sum and scale to -100 to 100 range\n",
+    "for i in range(1, size):\n",
+    "    counter[i] = counter[i-1] + counter_steps[i-1]\n",
+    "    rollout[i] = rollout[i-1] + rollout_steps[i-1]\n",
+    "    square[i] = square[i-1] + square_steps[i-1]\n",
+    "\n",
+    "# normalize values between -100 and 100, lowest value is -100, highest value is 100\n",
+    "counter = (counter - min(counter)) / (max(counter) - min(counter)) * 200 - 100\n",
+    "rollout = (rollout - min(rollout)) / (max(rollout) - min(rollout)) * 200 - 100\n",
+    "square = (square - min(square)) / (max(square) - min(square)) * 200 - 100\n",
+    "\n",
+    "# Create DataFrame\n",
+    "df = pd.DataFrame({\n",
+    "    'Counter': counter,\n",
+    "    'Rollout': rollout,\n",
+    "    'Square': square\n",
+    "})\n",
+    "\n",
+    "# add a timestamp column\n",
+    "df['Timestamp'] = pd.date_range(start='2025-01-01', periods=size, freq='5s')\n",
+    "\n",
+    "# Save to CSV\n",
+    "df.to_csv('random_walks_demo.csv', index=False)\n",
+    "\n",
+    "# Display first few rows\n",
+    "print(df.head())\n"
   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "e7c8eeb1",
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/html": [
+       "<div>\n",
+       "<style scoped>\n",
+       "    .dataframe tbody tr th:only-of-type {\n",
+       "        vertical-align: middle;\n",
+       "    }\n",
+       "\n",
+       "    .dataframe tbody tr th {\n",
+       "        vertical-align: top;\n",
+       "    }\n",
+       "\n",
+       "    .dataframe thead th {\n",
+       "        text-align: right;\n",
+       "    }\n",
+       "</style>\n",
+       "<table border=\"1\" class=\"dataframe\">\n",
+       "  <thead>\n",
+       "    <tr style=\"text-align: right;\">\n",
+       "      <th></th>\n",
+       "      <th>a</th>\n",
+       "      <th>b</th>\n",
+       "    </tr>\n",
+       "    <tr>\n",
+       "      <th>timestamp</th>\n",
+       "      <th></th>\n",
+       "      <th></th>\n",
+       "    </tr>\n",
+       "  </thead>\n",
+       "  <tbody>\n",
+       "    <tr>\n",
+       "      <th>2025-01-01</th>\n",
+       "      <td>1</td>\n",
+       "      <td>4</td>\n",
+       "    </tr>\n",
+       "    <tr>\n",
+       "      <th>2025-01-02</th>\n",
+       "      <td>2</td>\n",
+       "      <td>5</td>\n",
+       "    </tr>\n",
+       "    <tr>\n",
+       "      <th>2025-01-03</th>\n",
+       "      <td>3</td>\n",
+       "      <td>6</td>\n",
+       "    </tr>\n",
+       "  </tbody>\n",
+       "</table>\n",
+       "</div>"
+      ],
+      "text/plain": [
+       "            a  b\n",
+       "timestamp       \n",
+       "2025-01-01  1  4\n",
+       "2025-01-02  2  5\n",
+       "2025-01-03  3  6"
+      ]
+     },
+     "metadata": {},
+     "output_type": "display_data"
+    },
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "True\n"
+     ]
+    },
+    {
+     "data": {
+      "text/html": [
+       "<div>\n",
+       "<style scoped>\n",
+       "    .dataframe tbody tr th:only-of-type {\n",
+       "        vertical-align: middle;\n",
+       "    }\n",
+       "\n",
+       "    .dataframe tbody tr th {\n",
+       "        vertical-align: top;\n",
+       "    }\n",
+       "\n",
+       "    .dataframe thead th {\n",
+       "        text-align: right;\n",
+       "    }\n",
+       "</style>\n",
+       "<table border=\"1\" class=\"dataframe\">\n",
+       "  <thead>\n",
+       "    <tr style=\"text-align: right;\">\n",
+       "      <th></th>\n",
+       "      <th>a</th>\n",
+       "      <th>b</th>\n",
+       "    </tr>\n",
+       "    <tr>\n",
+       "      <th>timestamp</th>\n",
+       "      <th></th>\n",
+       "      <th></th>\n",
+       "    </tr>\n",
+       "  </thead>\n",
+       "  <tbody>\n",
+       "    <tr>\n",
+       "      <th>2025-01-03</th>\n",
+       "      <td>3</td>\n",
+       "      <td>6</td>\n",
+       "    </tr>\n",
+       "  </tbody>\n",
+       "</table>\n",
+       "</div>"
+      ],
+      "text/plain": [
+       "            a  b\n",
+       "timestamp       \n",
+       "2025-01-03  3  6"
+      ]
+     },
+     "metadata": {},
+     "output_type": "display_data"
+    },
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "3\n",
+      "1\n"
+     ]
+    }
+   ],
+   "source": [
+    "from pandas import DataFrame\n",
+    "\n",
+    "a = DataFrame({\n",
+    "    \"a\": {\"2025-01-01\": 1, \"2025-01-02\": 2, \"2025-01-03\": 3},\n",
+    "    \"b\": {\"2025-01-01\": 4, \"2025-01-02\": 5, \"2025-01-03\": 6},\n",
+    "})\n",
+    "\n",
+    "a.index.name = \"timestamp\"\n",
+    "\n",
+    "display(a)\n",
+    "\n",
+    "print(\"a\" in a.columns)\n",
+    "\n",
+    "b = a.tail(1)\n",
+    "\n",
+    "display(b)\n",
+    "\n",
+    "print(len(a))\n",
+    "print(len(b))\n",
+    "print(b.size)\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f3374174",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "<class 'str'>\n",
+      "All elements in index are of the same type\n"
+     ]
+    }
+   ],
+   "source": [
+    "from pandas import DataFrame\n",
+    "\n",
+    "data = DataFrame({\n",
+    "    \"a\": {\"2025-01-01\": 1, \"2025-01-02\": 2, \"2025-01-03\": 3},\n",
+    "    \"b\": {\"2025-01-01\": 4, \"2025-01-02\": 5, \"2025-01-03\": 6},\n",
+    "})\n",
+    "\n",
+    "index = data.index\n",
+    "\n",
+    "# Get type of first element of index\n",
+    "index_type = type(index[0])\n",
+    "\n",
+    "print(index_type)\n",
+    "\n",
+    "# Check if all in index are of the same type\n",
+    "if all(isinstance(i, index_type) for i in index):\n",
+    "    print(\"All elements in index are of the same type\")\n",
+    "else:\n",
+    "    print(\"Elements in index are of different types\")\n",
+    "\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "40e72c60",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "1fbb3788",
+   "metadata": {},
+   "outputs": [],
+   "source": []
  }
 ],
 "metadata": {
--- a/tests/laborious/activities/test_gates.py
+++ b/tests/laborious/activities/test_gates.py
@@ -379,8 +379,14 @@ async def test_format_prediction_no_timestamp(gates_activity):
    # Arrange
    input_data = {
        **metadata,
-        'data': {'prediction': [1], 'response_time': [0.1]},
-        'timestamp': '2023-05-26 11:12:27',
+        'data': {
+            'prediction': {
+                '2023-05-26 11:12:27': 1
+            },
+            'response_time': {
+                '2023-05-26 11:12:27': 0.1
+            }
+        },
        'model_id': 'test_model',
        'prediction_confidence': 0.9,
        'prediction_store_policy': 'lts:1'
@@ -404,9 +410,18 @@ async def test_format_prediction_with_timestamp_erl(gates_activity):
    # Arrange
    input_data = {
        **metadata,
-        'data': {'prediction': [1, 2, 3],
-                 'response_time': [0.1, 0.2, 0.3],
-                 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
+        'data': {
+            'prediction': {
+                '2023-05-26 11:12:27': 1,
+                '2023-05-26 11:12:28': 2,
+                '2023-05-26 11:12:29': 3,
+            },
+            'response_time': {
+                '2023-05-26 11:12:27': 0.1,
+                '2023-05-26 11:12:28': 0.2,
+                '2023-05-26 11:12:29': 0.3,
+            }
+        },
        'model_id': 'test_model',
        'prediction_confidence': 0.9,
        'prediction_store_policy': 'erl:2'
@@ -431,9 +446,18 @@ async def test_format_prediction_with_timestamp_lts(gates_activity):
    # Arrange
    input_data = {
        **metadata,
-        'data': {'prediction': [1, 2, 3],
-                 'response_time': [0.1, 0.2, 0.3],
-                 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
+        'data': {
+            'prediction': {
+                '2023-05-26 11:12:27': 1,
+                '2023-05-26 11:12:28': 2,
+                '2023-05-26 11:12:29': 3,
+            },
+            'response_time': {
+                '2023-05-26 11:12:27': 0.1,
+                '2023-05-26 11:12:28': 0.2,
+                '2023-05-26 11:12:29': 0.3,
+            }
+        },
        'model_id': 'test_model',
        'prediction_confidence': 0.9,
        'prediction_store_policy': 'lts:2'
--- a/tests/laborious/activities/test_mlflow.py
+++ b/tests/laborious/activities/test_mlflow.py
@@ -1,8 +1,10 @@
+from datetime import datetime
 from unittest.mock import ANY, MagicMock, patch

 import numpy as np
-from pandas import DataFrame
-from pytest import fixture, mark
+from pandas import DataFrame, Timestamp
+from pytest import fixture, mark, raises
+from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ
 from laborious.activities.mlflow import MLFlow
 from sientia_do.notifications.models import NotificationLevel

@@ -58,7 +60,7 @@ metadata = {
@mark.asyncio
@patch("laborious.activities.mlflow.DataFrame")
@patch("laborious.activities.mlflow.max")
-async def test_request_transform(mock_max, mock_dataframe, mlflow):
+async def test_request_transform_success(mock_max, mock_dataframe, mlflow):
    mock_max.return_value = '2024-01-02'
    # Mock input data
    input_data = {
@@ -78,7 +80,7 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow):
                'value': 1.0, 'created_at': '2024-01-01 12:00:00'}
        ],
        'model_name': 'test_model',
-        'model_retention': 30
+        'model_config': {}
    }

    # Mock the transform response
@@ -107,26 +109,35 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow):

    # Verify the repository was called with correct arguments
    mlflow.model_monitoring_repository.transform.assert_called_once_with(
-        'test_model', mock_dataframe, 30
+        'test_model', mock_dataframe, {}, metadata['metadata']
    )


@mark.asyncio
@patch("laborious.activities.mlflow.DataFrame")
+@patch("laborious.activities.mlflow.to_datetime")
@patch("laborious.activities.mlflow.max")
-async def test_request_predict(mock_max, mock_dataframe, mlflow):
+async def test_request_predict(mock_max, mock_to_datetime, mock_dataframe, mlflow):
    mock_max.return_value = '2024-01-02'
    # Mock input data
    input_data = {
        **metadata,
-        'data': [
-            {'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0},
-            {'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0},
-            {'timestamp': '2024-01-02', 'variable': 'var1', 'value': 3.0},
-            {'timestamp': '2024-01-02', 'variable': 'var2', 'value': 4.0}
-        ],
+        'data': {
+            "variable": {
+                "2024-01-01": "var1",
+                "2024-01-02": "var2",
+                "2024-01-03": "var1",
+                "2024-01-04": "var2"
+            },
+            "value": {
+                "2024-01-01": 1.0,
+                "2024-01-02": 2.0,
+                "2024-01-03": 3.0,
+                "2024-01-04": 4.0
+            }
+        },
        'model_name': 'test_model',
-        'model_retention': 30
+        'model_config': {}
    }

    # Mock the predict response
@@ -140,13 +151,23 @@ async def test_request_predict(mock_max, mock_dataframe, mlflow):
    mock_dataframe.return_value.replace.assert_called_once_with(
        np.nan, None, inplace=True
    )
+    mock_dataframe.return_value.__setitem__.assert_any_call(
+        'timestamp', mock_to_datetime.return_value.dt.strftime.return_value
+    )
+
+    mock_to_datetime.assert_called_once_with(
+        mock_dataframe.return_value.__getitem__.return_value, format=DATETIME_FORMAT_WITH_TZ
+    )
+    mock_to_datetime.return_value.dt.strftime.assert_called_once_with(
+        DATETIME_FORMAT
+    )

    # Verify the response
    assert response_data == expected_response

    # Verify the repository was called with correct arguments
    mlflow.model_monitoring_repository.predict.assert_called_once_with(
-        'test_model', mock_dataframe.return_value, 30
+        'test_model', mock_dataframe.return_value, {}, metadata['metadata']
    )


--- a/tests/laborious/utils/repository/test_model_repository.py
+++ b/tests/laborious/utils/repository/test_model_repository.py
@@ -2,7 +2,8 @@ from unittest.mock import ANY, MagicMock, call, patch
 import numpy as np
 from pandas import DataFrame
 import pytest
-from laborious.utils.repository import model_repository
+from datetime import datetime, timezone
+from pandas import Timestamp
 from laborious.utils.repository.model_repository import MLFlowRepository


@@ -22,32 +23,132 @@ def mlflow_repository():
        return repo


+metadata = {
+    "metadata": {
+        "model_id": "test_model",
+        "model_name": "test_model",
+        "workflow_name": "test_workflow",
+        "schema_name": "test_schedule",
+    },
+}
+
+
+class Any:
+    pass
+
+
+invalid_cases = [
+    (
+        {
+            'value': {
+                '2024-01-01 12:00:00': 1,
+                2024: 2
+            }
+        }
+    ),
+    (
+        {
+            'value': {
+                '2024-01-01': 1,
+                '2024-01-02': 2
+            }
+        }
+    ),
+    (
+        {
+            'value': {
+                Any(): 1,
+                Any(): 2
+            }
+        }
+    )
+]
+
+
+@pytest.mark.parametrize("data", invalid_cases)
+def test_detect_and_parse_datetime_index_error_cases(mlflow_repository, data):
+    input_data = DataFrame(
+        data
+    )
+
+    with pytest.raises(ValueError) as e:
+        mlflow_repository.detect_and_parse_datetime_index(
+            input_data, metadata['metadata'])
+
+        assert str(e) == "Index must be all timestamp like column. Valid formats are: pandas Timestamp, datetime, string in format %Y-%m-%d %H:%M:%S"
+
+
+valid_cases = [
+    (
+        {
+            'value': {
+                '2024-01-01 12:00:00+0000': 1,
+                '2024-01-02 12:00:00+0000': 2
+            }
+        }, ['2024-01-01 12:00:00+0000', '2024-01-02 12:00:00+0000']
+    ),
+    (
+        {
+            'value': {
+                datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc): 1,
+                datetime(2025, 1, 2, 12, 0, 0, tzinfo=timezone.utc): 2
+            }
+        }, ['2025-01-01 12:00:00+0000', '2025-01-02 12:00:00+0000']
+    ),
+    (
+        {
+            'value': {
+                Timestamp(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc): 1,
+                Timestamp(2026, 1, 2, 12, 0, 0, tzinfo=timezone.utc): 2
+            }
+        }, ['2026-01-01 12:00:00+0000', '2026-01-02 12:00:00+0000']
+    ),
+]
+
+
+@pytest.mark.parametrize("data,expected", valid_cases)
+def test_detect_and_parse_datetime_index_valid_format(mlflow_repository, data, expected):
+    input_data = DataFrame(data)
+
+    response = mlflow_repository.detect_and_parse_datetime_index(
+        input_data, metadata['metadata'])
+
+    assert response.index.tolist() == expected
+
+
 def test_transform_success(mlflow_repository):
-    data = 'data'
+    data = MagicMock()
    model_name = 'model'

-    output = mlflow_repository.transform(model_name, data, 1)
+    mlflow_repository.detect_and_parse_datetime_index = MagicMock()
+
+    output = mlflow_repository.transform(
+        model_name, data, {}, metadata['metadata'])

    mlflow_repository.model_serving.get_cached_transform.assert_called_once_with(
-        model_name, data, 1)
+        model_name, data, 0, 'sklearn', False, 'model', 'predict')
+
+    mlflow_repository.detect_and_parse_datetime_index.assert_called_once_with(
+        mlflow_repository.model_serving.get_cached_transform.return_value, metadata['metadata'])

    assert output == {
        'success': True,
-        'content': mlflow_repository.model_serving.get_cached_transform.return_value.to_dict.return_value
+        'content': mlflow_repository.detect_and_parse_datetime_index.return_value.to_dict.return_value
    }


 def test_transform_error(mlflow_repository):
-    data = 'data'
+    data = MagicMock()
    model_name = 'model'

    mlflow_repository.model_serving.get_cached_transform.side_effect = Exception(
        'error')

-    output = mlflow_repository.transform(model_name, data, 1)
+    output = mlflow_repository.transform(
+        model_name, data, {}, metadata['metadata'])

    mlflow_repository.model_serving.get_cached_transform.assert_called_once_with(
-        model_name, data, 1)
+        model_name, data, 0, 'sklearn', False, 'model', 'predict')

    assert output == {
        'success': False,
@@ -59,36 +160,53 @@ def test_transform_error(mlflow_repository):


 def test_predict_success(mlflow_repository):
-    data = 'data'
+    data = DataFrame({
+        'feat_1': {
+            'index_1': 2,
+            'index_2': 3
+        }
+    })
    model_name = 'model'
    mlflow_repository.model_serving.get_cached_predict.return_value = np.array(
        [2, 3]
    )

-    output = mlflow_repository.predict(model_name, data, 1)
+    output = mlflow_repository.predict(
+        model_name, data, {}, metadata['metadata'])

    mlflow_repository.model_serving.get_cached_predict.assert_called_once_with(
-        model_name, data, 1)
+        model_name, data, 0, 'pyfunc', False, 'model')

    assert output['success'] is True
-    assert output['content'] == {'prediction': {
-        0: 2,
-        1: 3
-    }, 'response_time': ANY}
+    assert output['content'] == {
+        'prediction': {
+            'index_1': 2,
+            'index_2': 3
+        }, 'response_time': {
+            'index_1': ANY,
+            'index_2': ANY
+        }
+    }


 def test_predict_error(mlflow_repository):
-    data = 'data'
+    data = DataFrame({
+        'feat_1': {
+            'index_1': 2,
+            'index_2': 3
+        }
+    })
    model_name = 'model'

    mlflow_repository.model_serving.get_cached_predict = MagicMock(
        side_effect=Exception('error')
    )

-    output = mlflow_repository.predict(model_name, data, 1)
+    output = mlflow_repository.predict(
+        model_name, data, {}, metadata['metadata'])

    mlflow_repository.model_serving.get_cached_predict.assert_called_once_with(
-        model_name, data, 1)
+        model_name, data, 0, 'pyfunc', False, 'model')

    assert output == {
        'success': False,
--- a/tests/laborious/workflows/subworkflows/test_prediction_process.py
+++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py
@@ -34,9 +34,12 @@ async def test_run(workflow_mock, prediction_process):
        'mlflow_transform_filters': {'test': 'filter'},
        'mlflow_predict_filters': {'test': 'filter'},
        'model_name': 'test_model_name',
-        'model_retention': '30',
+        'model_config': {
+            'retention': '30'
+        },
        'path_priority': ['continue', 'repeat', 'stop'],
        'opc_output_config': {'test': 'config'},
+        'prediction_store_policy': 'lts:1'
    }

    # Mock the activity responses
@@ -61,54 +64,54 @@ async def test_run(workflow_mock, prediction_process):

    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.get_last_timestamp, {
+            **metadata,
            'data': input_data['data'],
-            **metadata
        },
            retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.input_gate, {
+            **metadata,
            'filters': input_data['input_filters'],
            'data': input_data['data'],
            'path_priority': input_data['path_priority'],
-            **metadata
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.request_transform, {
+            **metadata,
            'data': input_data['data'],
            'model_name': input_data['model_name'],
-            'model_retention': input_data['model_retention'],
-            **metadata
+            'model_config': input_data['model_config'],
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.mlflow_response_gate, {
+            **metadata,
            'filters': input_data['mlflow_transform_filters'],
            'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'},
            'type': 'transform',
            'path_priority': input_data['path_priority'],
-            **metadata
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.mlflow_content_gate, {
+            **metadata,
            'filters': input_data['mlflow_transform_filters'],
            'data': 'transformed_data',
            'type': 'transform',
            'path_priority': input_data['path_priority'],
-            **metadata
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.request_predict, {
+            **metadata,
            'data': 'transformed_data',
            'model_name': input_data['model_name'],
-            'model_retention': input_data['model_retention'],
-            **metadata
+            'model_config': input_data['model_config'],
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.mlflow_response_gate, {
+            **metadata,
            'filters': input_data['mlflow_predict_filters'],
            'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'},
            'type': 'predict',
            'path_priority': input_data['path_priority'],
-            **metadata
        }, retry_policy=ANY, start_to_close_timeout=ANY)])

    workflow_mock.execute_child_workflow.assert_called_once_with(
@@ -121,11 +124,12 @@ async def test_run(workflow_mock, prediction_process):
            'timestamp': '2024-01-01',
            'model_id': 1,
            'model_name': 'test_model_name',
-            'model_retention': '30',
+            'model_config': input_data['model_config'],
            'opc_output_config': input_data['opc_output_config'],
            'schema': input_data['schema'],
            'table_name': input_data['table_name'],
-            'comment': 'Error'
+            'comment': 'Error',
+            'prediction_store_policy': input_data['prediction_store_policy']
        }
    )

@@ -145,7 +149,9 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process):
        'mlflow_transform_filters': {'test': 'filter'},
        'mlflow_predict_filters': {'test': 'filter'},
        'model_name': 'test_model_name',
-        'model_retention': '30',
+        'model_config': {
+            'retention': '30'
+        },
        'path_priority': ['continue', 'repeat', 'stop'],
        'opc_output_config': {'test': 'config'}
    }
@@ -164,13 +170,13 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process):
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.get_last_timestamp, {
             'data': input_data['data'],
-             **metadata
+             **metadata,
             }, retry_policy=ANY, start_to_close_timeout=ANY),
        call(Activities.input_gate, {
             'filters': input_data['input_filters'],
             'data': input_data['data'],
             'path_priority': input_data['path_priority'],
-             **metadata
+             **metadata,
             }, retry_policy=ANY, start_to_close_timeout=ANY)
    ])
    workflow_mock.execute_child_workflow.assert_not_called()
@@ -191,7 +197,9 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
        'mlflow_transform_filters': {'test': 'filter'},
        'mlflow_predict_filters': {'test': 'filter'},
        'model_name': 'test_model_name',
-        'model_retention': '30',
+        'model_config': {
+            'retention': '30'
+        },
        'path_priority': ['continue', 'repeat', 'stop'],
        'opc_output_config': {'test': 'config'}
    }
@@ -212,7 +220,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.get_last_timestamp, {
             'data': input_data['data'],
-             **metadata
+             **metadata,
             },
             retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
@@ -220,14 +228,14 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
             'filters': input_data['input_filters'],
             'data': input_data['data'],
             'path_priority': input_data['path_priority'],
-             **metadata
+             **metadata,
             },
             retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.request_transform, {
            'data': input_data['data'],
            'model_name': input_data['model_name'],
-            'model_retention': input_data['model_retention'],
+            'model_config': input_data['model_config'],
            **metadata
        },
            retry_policy=ANY, start_to_close_timeout=ANY)
@@ -260,7 +268,9 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
        'mlflow_transform_filters': {'test': 'filter'},
        'mlflow_predict_filters': {'test': 'filter'},
        'model_name': 'test_model_name',
-        'model_retention': '30',
+        'model_config': {
+            'retention': '30'
+        },
        'path_priority': ['continue', 'repeat', 'stop'],
        'opc_output_config': {'test': 'config'}
    }
@@ -285,7 +295,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.get_last_timestamp, {
             'data': input_data['data'],
-             **metadata
+             **metadata,
             },
             retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
@@ -293,14 +303,14 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
             'filters': input_data['input_filters'],
             'data': input_data['data'],
             'path_priority': input_data['path_priority'],
-             **metadata
+             **metadata,
             },
             retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.request_transform, {
            'data': input_data['data'],
            'model_name': input_data['model_name'],
-            'model_retention': input_data['model_retention'],
+            'model_config': input_data['model_config'],
            **metadata
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
@@ -309,7 +319,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
            'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'},
            'type': 'transform',
            'path_priority': input_data['path_priority'],
-            **metadata
+            **metadata,
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.mlflow_content_gate, {
@@ -317,7 +327,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
            'data': 'transformed_data',
            'type': 'transform',
            'path_priority': input_data['path_priority'],
-            **metadata
+            **metadata,
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_child_workflow.assert_not_called()

@@ -338,7 +348,9 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
        'mlflow_transform_filters': {'test': 'filter'},
        'mlflow_predict_filters': {'test': 'filter'},
        'model_name': 'test_model_name',
-        'model_retention': '30',
+        'model_config': {
+            'retention': '30'
+        },
        'path_priority': ['continue', 'repeat', 'stop'],
        'opc_output_config': {'test': 'config'}
    }
@@ -364,7 +376,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.get_last_timestamp, {
             'data': input_data['data'],
-             **metadata
+             **metadata,
             },
             retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
@@ -372,14 +384,14 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
             'filters': input_data['input_filters'],
             'data': input_data['data'],
             'path_priority': input_data['path_priority'],
-             **metadata
+             **metadata,
             },
             retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.request_transform, {
            'data': input_data['data'],
            'model_name': input_data['model_name'],
-            'model_retention': input_data['model_retention'],
+            'model_config': input_data['model_config'],
            **metadata
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
@@ -388,7 +400,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
            'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'},
            'type': 'transform',
            'path_priority': input_data['path_priority'],
-            **metadata
+            **metadata,
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.mlflow_content_gate, {
@@ -396,13 +408,13 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
            'data': 'transformed_data',
            'type': 'transform',
            'path_priority': input_data['path_priority'],
-            **metadata
+            **metadata,
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
        call(Activities.request_predict, {
            'data': 'transformed_data',
            'model_name': input_data['model_name'],
-            'model_retention': input_data['model_retention'],
+            'model_config': input_data['model_config'],
            **metadata
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_local_activity_method.assert_has_calls([
@@ -411,7 +423,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
            'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'},
            'type': 'predict',
            'path_priority': input_data['path_priority'],
-            **metadata
+            **metadata,
        }, retry_policy=ANY, start_to_close_timeout=ANY)])
    workflow_mock.execute_child_workflow.assert_not_called()

@@ -428,7 +440,9 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
    model = 'test_model'
    last_timestamp = '2024-01-01'
    model_name = 'test_model_name'
-    model_retention = '30'
+    model_config = {
+        'retention': '30'
+    }

    # Act
    result = await prediction_process.path_flag_handler(
@@ -439,7 +453,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
            'model_id': model,
            'last_timestamp': last_timestamp,
            'model_name': model_name,
-            'model_retention': model_retention
+            'model_config': model_config
        }, confidence, last_timestamp, ""
    )

@@ -461,7 +475,9 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
    model = 'test_model'
    last_timestamp = '2024-01-01'
    model_name = 'test_model_name'
-    model_retention = '30'
+    model_config = {
+        'retention': '30'
+    }

    # Act
    result = await prediction_process.path_flag_handler(
@@ -472,7 +488,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
            'model_id': model,
            'last_timestamp': last_timestamp,
            'model_name': model_name,
-            'model_retention': model_retention
+            'model_config': model_config
        }, confidence, last_timestamp, ""
    )

@@ -505,7 +521,10 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
    model = 'test_model'
    last_timestamp = '2024-01-01'
    model_name = 'test_model_name'
-    model_retention = '30'
+    model_config = {
+        'retention': '30'
+    }
+    prediction_store_policy = 'erl:1'

    # Act
    result = await prediction_process.path_flag_handler(
@@ -516,8 +535,9 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
            'model_id': model,
            'last_timestamp': last_timestamp,
            'model_name': model_name,
-            'model_retention': model_retention,
-            'opc_output_config': {'test': 'config'}
+            'model_config': model_config,
+            'opc_output_config': {'test': 'config'},
+            'prediction_store_policy': prediction_store_policy
        }, confidence, last_timestamp, 'Prediction Process'
    )

@@ -534,11 +554,12 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
            'timestamp': last_timestamp,
            'model_id': model,
            'model_name': model_name,
-            'model_retention': model_retention,
+            'model_config': model_config,
            'schema': schema,
            'table_name': table_name,
            'comment': 'Prediction Process',
-            'opc_output_config': {'test': 'config'}
+            'opc_output_config': {'test': 'config'},
+            'prediction_store_policy': prediction_store_policy
        }
    )

@@ -555,8 +576,10 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
    model = 'test_model'
    last_timestamp = '2024-01-01'
    model_name = 'test_model_name'
-    model_retention = '30'
-
+    model_config = {
+        'retention': '30'
+    }
+    prediction_store_policy = 'erl:1'
    # Act
    result = await prediction_process.path_flag_handler(
        data, path_flag, {
@@ -566,8 +589,9 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
            'model_id': model,
            'last_timestamp': last_timestamp,
            'model_name': model_name,
-            'model_retention': model_retention,
-            'opc_output_config': {'test': 'config'}
+            'model_config': model_config,
+            'opc_output_config': {'test': 'config'},
+            'prediction_store_policy': prediction_store_policy
        }, confidence, last_timestamp, ""
    )

--- a/tests/laborious/workflows/test_predictions_batch.py
+++ b/tests/laborious/workflows/test_predictions_batch.py
@@ -33,7 +33,11 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch
        'schema': 'test_schema',
        'table_name': 'test_table',
        'opc_output_config': 'test_opc_output_config',
-        'datetime_columns': ['timestamp', 'created_at']
+        'datetime_columns': ['timestamp', 'created_at'],
+        'prediction_store_policy': 'erl:1',
+        'model_config': {
+            'retention': '30'
+        }
    }

    await predictions_batch.run(input_data)
@@ -72,9 +76,10 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch
                'POLICY': 'STOP'
            }
        }),
-        'model_retention': input_data.get('model_retention', 60),
+        'model_config': input_data.get('model_config', {}),
        'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']),
-        'opc_output_config': input_data.get('opc_output_config', {})
+        'opc_output_config': input_data.get('opc_output_config', {}),
+        'prediction_store_policy': input_data.get('prediction_store_policy', 'erl:1')
    }

    workflow_mock.execute_child_workflow.assert_has_calls([
--- a/values.yaml
+++ b/values.yaml
@@ -7,13 +7,13 @@ replicaCount: 1

 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
 image:
-  repository: aignosi.azurecr.io/sientia-module
+  repository: aignosi.azurecr.io/sientia-module-courier
  # This sets the pull policy for images.
  pullPolicy: Always
  # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.4.4"
+  tag: "0.0.2"

-# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
+0# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
 - name: docker-hub-secret
 # This is to override the chart name.
@@ -151,7 +151,7 @@ env:
  - name: GITHUB_REPO_URL
    value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
  - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1205-alterar-opc-para-assincrono"
+    value: SIENTIAPDE-1222-ajustar-a-library-para-fazer-o-download-do-courier
  - name: PYTHON_APP
    value: "laborious.worker.worker"

@@ -178,7 +178,7 @@ env:
  - name: MLFLOW_USERNAME
    value: "aignosi"
  - name: MLFLOW_PASSWORD
-    value: "aignosi"
+    value: "1L0FP50j3ncp123"

  - name: OPC_ID
    value: "1"
Author	SHA1	Message	Date
vitor-aignosi	58f8cb9720	SIENTIAPDE-1222 Update .gitignore to include 'tmp/' directory and ensure '.env' is listed - Added 'tmp/' to the .gitignore file to prevent temporary files from being tracked. - Confirmed that '.env' is included to avoid committing sensitive environment variables.	2025-09-24 10:33:18 -03:00
vitor-aignosi	e0720eebf1	SIENTIAPDE-1222 Remove deprecated files and configurations from transformer_pyfunc module - Deleted conda.yaml, MLmodel, python_env.yaml, requirements.txt, and various utility scripts related to data processing and model handling. - Removed binary files including python_model.pkl and training_transformer.pkl to clean up the artifacts directory. - This cleanup is part of the effort to streamline the transformer_pyfunc module and eliminate unused components.	2025-09-24 10:32:59 -03:00
vitor-aignosi	4ebbb97de9	Update tests/laborious/activities/test_mlflow.py Co-authored-by: codeant-ai[bot] <151821869+codeant-ai[bot]@users.noreply.github.com>	2025-09-24 10:30:23 -03:00
vitor-aignosi	d6c34dcaf7	SIENTIAPDE-1222 Add placeholder class 'Any' in test_model_repository.py and update invalid_cases to use it - Introduced a new placeholder class 'Any' to be used in test cases. - Updated the 'invalid_cases' list to replace integer keys with instances of the 'Any' class, enhancing test coverage for key types.	2025-09-24 09:59:26 -03:00
vitor-aignosi	84c1371d6c	SIENTIAPDE-1222 Refactor model configuration handling in MLFlow and workflows - Replaced 'model_retention' with 'model_config' to encapsulate retention settings and improve consistency across various components. - Updated test cases to reflect changes in argument structure, ensuring compatibility with the new model configuration format. - Added 'prediction_store_policy' to input data handling in workflows for enhanced configuration management.	2025-09-17 16:02:54 -03:00
vitor-aignosi	d6fdbc58bb	SIENTIAPDE-1222 Enhance MLFlow data handling by adding timestamp column and improving debug logging - Added a 'timestamp' column to the input data, converting the index to a datetime format for better tracking of predictions. - Improved debug logging to provide clearer context by including the input data preview in the log output.	2025-09-17 11:15:05 -03:00
vitor-aignosi	fc3c4ebb45	SIENTIAPDE-1222 Update MLFlow class to pass logger instance directly to MLFlowRepository - Modified the initialization of MLFlowRepository in the MLFlow class to pass the logger instance directly, improving logging capabilities and consistency across the application.	2025-09-17 10:27:02 -03:00
vitor-aignosi	863bbdcc57	SIENTIAPDE-1222 SIENTIAPDE-1222 Update MLFlow methods to include metadata parameter - Modified the transform and predict method calls in the MLFlow class to include a new 'metadata' parameter, enhancing the functionality and data handling capabilities of the model monitoring repository.	2025-09-17 10:11:54 -03:00
vitor-aignosi	b2ef523c3b	SIENTIAPDE-1222 Update test cases in model repository and prediction process - Replaced string data with MagicMock in test_transform_success and test_transform_error to improve test isolation. - Updated the predict method calls in test_predict_success and test_predict_error to reflect changes in argument structure. - Added 'prediction_store_policy' to the test_run configuration in test_prediction_process for consistency with recent updates.	2025-09-17 10:07:32 -03:00
vitor-aignosi	2075b30243	SIENTIAPDE-1222 Refactor datetime index handling in MLFlow and MLFlowRepository - Moved the detect_and_parse_datetime_index method from MLFlow to MLFlowRepository for better organization and reusability. - Updated the method to include enhanced logging and error handling for invalid datetime formats. - Adjusted the transform method in MLFlowRepository to utilize the new datetime index parsing logic. - Added unit tests for both valid and invalid datetime index cases to ensure robustness.	2025-09-17 08:59:12 -03:00
vitor-aignosi	25bc4d06ca	SIENTIAPDE-1222 Update README.md to reflect new features and configuration changes - Added details about two dedicated task queues: `predictions_batch-queue` and `minimal_retrain-queue`. - Enhanced descriptions of activities and workflows, including multiple inheritance patterns and configurable MLFlow model serving. - Updated monitoring metrics section to include new labels and metrics for prediction and OPC export operations. - Revised configuration section with updated default values and added new environment variables for Kubernetes pod identification. - Improved clarity in the Predictions Batch Workflow configuration example, including structured input filters and updated retention policies.	2025-09-16 17:07:44 -03:00
vitor-aignosi	a5fc526f61	SIENTIAPDE-1222 Update model configuration key for compression in MLFlowRepository - Changed the key for compression from 'compressed' to 'is_compressed' in model configuration handling to ensure consistency with updated standards.	2025-09-16 13:04:43 -03:00
vitor-aignosi	2b45c1cbd4	SIENTIAPDE-1222 Update model configuration keys in MLFlowRepository for consistency - Changed 'model_retention' to 'retention_minutes' and 'is_compressed' to 'compressed' in model configuration handling, ensuring alignment with updated configuration standards.	2025-09-16 12:34:56 -03:00
vitor-aignosi	b1951ea2c0	SIENTIAPDE-1222 Enhance debug logging in MLFlowRepository to improve data traceability - Added a debug logging statement to capture received data for model prediction, improving visibility into input data. - Changed an existing info logging statement to debug level for consistency, ensuring all relevant data is logged at the appropriate level.	2025-09-16 11:54:35 -03:00
vitor-aignosi	74f7e6c024	SIENTIAPDE-1222 Update prediction_store_policy handling in workflows - Added 'prediction_store_policy' to the input data handling in PredictionsBatch, ensuring a default value of 'lts:1' is used when not provided. - Modified FormatAndExportPrediction to directly use 'prediction_store_policy' from input_data, removing the default fallback. - Updated PredictionProcess to include 'prediction_store_policy' in the output data structure, ensuring consistency across workflows.	2025-09-16 11:41:27 -03:00
vitor-aignosi	7fb416313e	SIENTIAPDE-1222 Enhance MLFlow debug logging to limit output size - Updated debug logging statements in the MLFlow activity to include a maximum of 5 items and a depth of 5 for the sample dictionary, improving readability and reducing log clutter for raw and transformed response data.	2025-09-16 10:36:43 -03:00
vitor-aignosi	ce288c3926	SIENTIAPDE-1222 Refactor debug logging in Gates activity for improved readability - Reformatted the debug logging statement for input data in the Gates activity to enhance clarity and maintain consistency with previous logging improvements.	2025-09-16 10:31:30 -03:00
vitor-aignosi	0ffd96e734	SIENTIAPDE-1222 Refactor MLFlow debug logging for improved readability - Reformatted debug logging statements in the MLFlow activity to enhance clarity and consistency. - Ensured that the output of raw and transformed response data is presented in a more readable format, maintaining the use of create_sample_dict for better visualization.	2025-09-16 10:29:35 -03:00
vitor-aignosi	eefd0815d8	SIENTIAPDE-1222 Update image tag in values.yaml and enhance debug logging in Gates and MLFlow activities - Updated the image tag in values.yaml from '0.0.1' to '0.0.2'. - Improved debug logging in the Gates activity to format input data and filters for better readability. - Enhanced MLFlow activity logging to include formatted output for raw and transformed response data, ensuring consistent logging format.	2025-09-16 10:22:51 -03:00
vitor-aignosi	41af0a8032	SIENTIAPDE-1222 Update requirements and enhance logging in Gates and MLFlow activities - Updated the sientia-dataops-library and sientia-mlops-library dependencies in requirements.txt to the latest versions. - Improved debug logging in the Gates activity to display a sample of input data and filters, enhancing clarity and reducing output size. - Refactored MLFlow activity logging to utilize the create_sample_dict function for better visualization of nested data structures in logs.	2025-09-16 08:54:27 -03:00
vitor-aignosi	1a9c1a31c5	SIENTIAPDE-1222 SIENTIAPDE-1222 Enhance MLFlow logging with sample dictionary for response data - Introduced a new method to create a sample dictionary for debugging, allowing for better visualization of nested data structures in logs. - Updated debug logging to utilize the new sampling method for raw and transformed response data, improving clarity and reducing output size. - Adjusted logging for processed input data to display only the first few rows, enhancing readability.	2025-09-16 08:16:30 -03:00
vitor-aignosi	6d10e4c597	SIENTIAPDE-1222 Refactor MLFlow logging to enhance data output clarity - Updated debug logging to use data.to_string() for processed input data, improving readability. - Modified raw response data logging to format the output as a string, ensuring consistent logging format.	2025-09-16 07:55:56 -03:00
vitor-aignosi	cc692e6dbd	SIENTIAPDE-1222 Update values.yaml and MLFlow logging for courier integration - Changed the image repository to 'sientia-module-courier' and updated the image tag to '0.0.1'. - Modified environment variables for GITHUB_BRANCH and MLFLOW_PASSWORD to reflect new configurations. - Enhanced MLFlow logging to include additional debug statements for raw response data and added a check for empty DataFrames.	2025-09-16 07:52:21 -03:00
vitor-aignosi	a817d9855b	SIENTIAPDE-1222 Refactor MLFlow logging to improve data output clarity - Updated the debug logging to directly capture the output of data.to_csv, enhancing traceability of processed input data. - Removed redundant debug statements for raw response data to streamline logging and focus on essential information.	2025-09-15 15:16:27 -03:00
vitor-aignosi	df7ce9e79b	SIENTIAPDE-1222 Remove obsolete data files: deleted data.csv and response_data.csv to streamline project structure and eliminate unused resources.	2025-09-15 14:47:46 -03:00
vitor-aignosi	934298b3c3	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.	2025-09-15 14:47:24 -03:00
vitor-aignosi	caca923717	SIENTIAPDE-1214 SIENTIAPDE-1214: Update requirements.txt to clarify dependencies and improve project setup - Commented out the previous sientia-mlops-library dependency for better clarity. - Ensured that the requirements.txt reflects the current state of dependencies for easier management.	2025-09-12 14:30:52 -03:00
vitor-aignosi	6df04b72e5	SIENTIAPDE-1214 SIENTIAPDE-1214: Refactor MLFlow and model repository methods to use model_config dictionary - Updated MLFlow class methods to accept model_config instead of model_retention for improved flexibility. - Modified model_repository methods to handle model_config, extracting necessary parameters for transformation and prediction. - Adjusted predictions_batch and prediction_process workflows to utilize model_config for better configuration management. - Commented out the previous sientia-mlops-library dependency in requirements.txt for clarity.	2025-09-12 14:23:35 -03:00
vitor-aignosi	4a4043c354	SIENTIAPDE-1214 Update README.md to include additional PostgreSQL, OPC, and MongoDB configuration options for enhanced clarity and usability	2025-09-05 16:49:38 -03:00
Bruno Domingues	8340189741	Merge pull request #20 from Aignosi/SIENTIAPDE-1182-ajustar-laborious-para-pegar-timestamp-da-resposta-do-mlflow SIENTIAPDE-1182: Enhance tests and refactor data structures for improved clarity and functionality	2025-09-05 10:39:41 -03:00
vitor-aignosi	f971e711bd	SIENTIAPDE-1182 Update tests.ipynb and gates.py for enhanced functionality and clarity - Adjusted execution count in tests.ipynb for reproducibility. - Modified DataFrame creation in tests.ipynb to dynamically set size and update timestamp frequency. - Added a new filter function for empty data in gates.py to improve data handling.	2025-09-04 15:39:01 -03:00
vitor-aignosi	8ab1b9b437	SIENTIAPDE-1182 Refactor test data structure in gates.py and model_repository.py for improved clarity and consistency - Updated test cases in test_gates.py to use dictionaries for prediction and response_time, associating values with timestamps. - Modified test_predict_success in test_model_repository.py to create a DataFrame with named indices for better readability in assertions.	2025-09-04 12:16:14 -03:00
vitor-aignosi	2e420dadba	SIENTIAPDE-1182 Enhance tests.ipynb and gates.py for improved logging and output clarity - Set execution count to null in tests.ipynb for reproducibility. - Added stdout output to tests.ipynb for better visibility of results. - Updated DataFrame display logic in tests.ipynb to include additional print statements for length and size. - Improved debug logging in gates.py to log prediction data as a string for better traceability. - Changed info log in gates.py to use len() instead of size for consistency.	2025-09-04 09:51:58 -03:00
vitor-aignosi	da21c128f8	SIENTIAPDE-1182 Update tests.ipynb and values.yaml for improved functionality and versioning - Updated execution count in tests.ipynb for reproducibility. - Modified DataFrame creation in tests.ipynb to include a timestamp column and save to CSV. - Changed image tag in values.yaml from "0.4.4" to "0.4.5" for versioning. - Updated GITHUB_BRANCH in values.yaml to reflect the latest branch adjustments.	2025-09-04 09:36:23 -03:00
Bruno Domingues	460bfe4073	Merge pull request #19 from Aignosi/SIENTIAPDE-1084-ajustar-documentacao SIENTIAPDE-1084: Refactor OPC activities, update prediction workflow, and enhance documentation	2025-09-03 13:31:38 +00:00
vitor-aignosi	456eb5c674	SIENTIAPDE-1084 Refactor debug logging in gates.py to remove f-string usage for improved consistency and readability	2025-09-03 09:53:07 -03:00
vitor-aignosi	69a44d5200	SIENTIAPDE-1084 Update requirements.txt to upgrade sientia-dataops-library from version 1.4.4 to 1.4.5 and sientia-mlops-library from version 0.38.12 to 0.38.13	2025-09-03 09:08:12 -03:00
vitor-aignosi	bac17579d6	SIENTIAPDE-1084 Refactor connectors_config.py and conditional_filters.py for improved configuration management and data filtering - Updated PostgreSQL and MLFlow configuration functions to enhance default values and environment variable handling. - Simplified OPC server configuration logic and improved MongoDB connection string construction. - Refactored conditional filters to streamline null value checks and empty data validation, removing unnecessary comments and examples for clarity. - Removed extensive module docstrings to enhance code readability.	2025-09-03 08:42:24 -03:00
vitor-aignosi	e0cb3cded7	SIENTIAPDE-1084 Remove requirements_prepared.txt as it is no longer needed in the project.	2025-09-03 08:20:14 -03:00