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.

.gitignore

@@ -43,4 +43,6 @@ git_key*
 
 git_log
 
-.env
+.env
+
+tmp/

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
+  - Labels: `pod_id`
-- `laborious_prediction_confidence_monitor`: Prediction confidence monitoring
-- `laborious_prediction_response_time_monitor`: Prediction response time monitoring
 
-### MLFlow Metrics
+### Prediction Operation Metrics
-- Model transformation and prediction success rates
+- `laborious_predictions_written_count`: Counter for successful prediction exports
-- API response times and error rates
+  - Labels: `pod_id`, `model_name`, `pipeline_name`
-- Model retention and versioning metrics
+- `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
+### OPC Export Metrics
-- PostgreSQL export operation counts and response times
+- `laborious_prediction_opc_writing_count`: Counter for OPC server write operations
-- OPC server write operations and performance
+  - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id`
-- Data quality filter pass/fail rates
+- `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 |
+| `POSTGRES_MIN_CONNECTIONS` | Minimum PostgreSQL connections | `5` | No |
-| `MLFLOW_PORT` | MLFlow server port | `5000` | Yes |
+| `POSTGRES_MAX_CONNECTIONS` | Maximum PostgreSQL connections | `20` | No |
-| `MLFLOW_USERNAME` | MLFlow username | `admin` | Yes |
+| `MLFLOW_HOST` | MLFlow server hostname | `http://localhost` | Yes |
-| `MLFLOW_PASSWORD` | MLFlow password | `admin` | 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
+  "frequency": "30s",
-  "max_retry_policy": 1, # Maximum number of retries for the workflow
+  "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",
+      "server_id": "server1",
-      "type": "prediction", # Type of tag to write, can be prediction or confidence
+      "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": [
+  "input_filters": {
-    {
+    "EMPTY_DATA": {"POLICY": "STOP"},
-      "filter_name": "EMPTY_DATA", # Required filter
+    "SPECIFIC_VARIABLES_NULL_VALUES": {
-      "policy": "STOP"
+      "POLICY": "CONTINUE",
-    },
+      "config": {"variables": ["Counter"]}
-    {
-      "filter_name": "SPECIFIC_VARIABLES_NULL_VALUES",
-      "policy": "CONTINUE",
-      "config": {
-        "variables": [
-          "Counter"
-        ]
-      }
     }
-  ],
+  },
-  "mlflow_transform_filters": [
+  "mlflow_transform_filters": {
-    {
+    "API_ERROR": {"POLICY": "REPEAT"},
-      "filter_name": "API_ERROR", # Required filter
+    "NAN_VALUES": {"POLICY": "STOP"}
-      "policy": "REPEAT"
+  },
-    },
+  "mlflow_predict_filters": {
-    {
+    "API_ERROR": {"POLICY": "CONTINUE"}
-      "filter_name": "NAN_VALUES",
+  },
-      "policy": "STOP"
+  "path_priority": ["STOP", "CONTINUE", "REPEAT"],
-    }
-  ],
-  "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"
-  ],
   "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

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"Input data:\n {data.head(5).to_string()}", metadata)
-        self.debug(f"Filters: {filters}", 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.info(
-            self.warning(
+            f"Sorting data by timestamp and applying policy: {policy_type}:{policy_value}", metadata)
-                "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)
 
-            # If policy_type is lts, we need to sort the data by timestamp descending and take the first policy_value rows
+        # 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':
+        if policy_type == 'lts':
-                self.debug(
+            self.debug(
-                    f"Sorting data by timestamp descending", metadata)
+                "Sorting data by timestamp descending", metadata)
-                data = data.sort_values(by='timestamp', ascending=False)
+            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
+        # 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':
+        elif policy_type == 'erl':
-                self.debug(
+            self.debug(
-                    f"Sorting data by timestamp ascending", metadata)
+                "Sorting data by timestamp ascending", metadata)
-                data = data.sort_values(by='timestamp', ascending=True)
+            data = data.sort_values(by='timestamp', ascending=True)
-            else:
+        else:
-                self.error(
+            self.error(
-                    f"Invalid policy type: {policy_type}, using default policy", metadata)
+                f"Invalid policy type: {policy_type}, using default policy", metadata)
-                raise ValueError(
+            raise ValueError(
-                    f"Invalid policy type: {policy_type}")
+                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.info(f"Prediction formatted: {len(data)} rows", metadata)
-        self.debug(f"Prediction data: {data.to_string()}", 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)

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(
-        self.debug(json.dumps(response_data, indent=4), metadata)
+            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(
-        self.debug(json.dumps(response_data, indent=4), metadata)
+            f"Prediction response data: \n {create_sample_dict(response_data, max_items=5, max_depth=5)}", metadata)
 
         self.info("Data predicted successfully", metadata)
 

laborious/utils/connectors_config.py

@@ -1,205 +1,99 @@
-"""
+from os import getenv
-Connectors Configuration Module
+import json
-
-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 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_MIN_CONNECTIONS: Minimum connection pool size (default: 5)
-        POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 10)
+        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'),
+        'host': getenv('POSTGRES_HOST', 'localhost'),
-        'port': int(os.getenv('POSTGRES_PORT', '5432')),
+        'port': int(getenv('POSTGRES_PORT', '5432')),
-        'user': os.getenv('POSTGRES_USER', 'sientia'),
+        'user': getenv('POSTGRES_USER', 'sientia'),
-        'password': os.getenv('POSTGRES_PASSWORD', 'sientia'),
+        'password': getenv('POSTGRES_PASSWORD', 'sientia'),
-        'dbname': os.getenv('POSTGRES_DBNAME', 'sientia'),
+        'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
-        'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '1')),
+        'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
-        'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '10'))
+        '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_HOST: MLFlow server hostname (default: http://localhost)
-        MLFLOW_PORT: MLFlow server port (default: 5000)
+        MLFLOW_PORT: MLFlow server port (default: 5080)
-        MLFLOW_USERNAME: MLFlow username (default: admin)
+        MLFLOW_USERNAME: MLFlow username (default: aignosi)
-        MLFLOW_PASSWORD: MLFlow password (default: admin)
+        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'),
+        'host': getenv('MLFLOW_HOST', 'http://localhost'),
-        'port': int(os.getenv('MLFLOW_PORT', '5000')),
+        'port': int(getenv('MLFLOW_PORT', '5080')),
-        'username': os.getenv('MLFLOW_USERNAME', 'admin'),
+        'username': getenv('MLFLOW_USERNAME', 'aignosi'),
-        'password': os.getenv('MLFLOW_PASSWORD', 'admin')
+        '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_ID: OPC server ID (fallback, default: 1)
-        OPC_NAME: Single OPC server name (fallback)
+        OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840)
-        OPC_SERVER_URI: Single OPC server URI (fallback)
+        OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840)
-        OPC_CERT_PATH: Client certificate path (fallback)
+        OPC_CERT_PATH: Client certificate path (fallback, default: None)
-        OPC_PRIVATE_KEY_PATH: Client private key path (fallback)
+        OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None)
-        OPC_SERVER_CERT_PATH: Server certificate path (fallback)
+        OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None)
-        OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback)
+        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_raw = getenv('OPC_CONFIG', None)
-    opc_config = os.getenv('OPC_CONFIG')
+
-    if opc_config:
+    if opc_raw:
-        try:
+        return json.loads(opc_raw)
-            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
     return {
-        'default': {
+        getenv('OPC_ID', '1'): {
-            'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'),
+            'id': getenv('OPC_ID', '1'),
-            'name': os.getenv('OPC_NAME', 'DefaultServer'),
+            'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'),
-            'server_uri': os.getenv('OPC_SERVER_URI', 'urn:default:opcua'),
+            'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
-            'cert_path': os.getenv('OPC_CERT_PATH', ''),
+            'cert_path': getenv('OPC_CERT_PATH', None),
-            'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', ''),
+            'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None),
-            'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', ''),
+            'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None),
-            'reconnection_interval': int(os.getenv('OPC_RECONNECTION_INTERVAL', '5000'))
+            '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_USERNAME: MongoDB username (default: root)
-        MONGODB_DATABASE: MongoDB database name (default: sientia)
+        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'),
+        'connection_string': connection_string,
-        'database_name': os.getenv('MONGODB_DATABASE', 'sientia')
+        'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
+        'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600
     }

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.
+    Filter to check if specific variables contain null values.
-    
+
-    This function checks if the input data meets minimum requirements for
+    This function examines a DataFrame to determine if any of the specified variables
-    processing. It can validate data size, completeness, and other quality
+    contain null (NaN) values. It returns True if null values are found for any of
-    metrics to ensure sufficient data is available for ML operations.
+    the specified variables, False otherwise.
-    
+
-    The filter implements multiple validation criteria:
-    1. Data frame size validation
-    2. Row count validation
-    3. Column completeness validation
-    4. Configurable threshold checking
-    
     Args:
-        data: Input data as pandas DataFrame
+        data (DataFrame): The pandas DataFrame to be examined. Must contain columns
-        config: Filter configuration dictionary
+            named 'variable' and 'value'.
-            Required keys:
+        config (dict): Configuration dictionary containing the following key:
-                - min_rows (int, optional): Minimum number of rows required
+            - variables (list): List of variable names to check for null values
-                - min_columns (int, optional): Minimum number of columns required
+
-                - min_data_points (int, optional): Minimum total data points required
-    
     Returns:
-        bool: True if data should be filtered (fails quality check), False otherwise
+        bool: True if any of the specified variables contain null values,
-        
+              False if none of the specified variables contain null values.
-    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)
     """
-    # Check if data is completely empty
+    return not data[
-    if data.empty:
+        data['variable'].isin(config['variables']) & data['value'].isna()].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
 
 
-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.
+    Filter to check if the DataFrame is empty.
-    
+
-    This function checks for null values in specified variables and determines
+    This function determines whether the provided DataFrame contains any data.
-    if the data quality is sufficient for processing. It can validate
+    It's a simple utility function that can be used in conditional logic to
-    individual columns or groups of columns for data completeness.
+    handle cases where no data is available.
-    
+
-    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
-    
     Args:
-        data: Input data as pandas DataFrame
+        data (DataFrame): The pandas DataFrame to be checked for emptiness.
-        config: Filter configuration dictionary
+        _config (dict): Configuration dictionary (unused in this function).
-            Required keys:
+            The underscore prefix indicates this parameter is required for
-                - variables (list): List of variable names to check
+            interface consistency but not used in the implementation.
-                - 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
-    
     Returns:
-        bool: True if data should be filtered (fails quality check), False otherwise
+        bool: True if the DataFrame is empty (has no rows), False if it contains data.
-        
+
-    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.
     """
-    # Get configuration
+    return data.empty
-    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

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(
+                'content': transformed_data.to_dict()
-                    model_name, data, model_retention).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 {

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

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': input_data['prediction_store_policy']
-                        'prediction_store_policy', 'lts:1')
                 },
                 retry_policy=retry_policy,
                 start_to_close_timeout=timedelta(seconds=60)

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

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-dataops-library.git@1.4.6
-git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.12
+git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0
 prometheus-client

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

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",
+      "     Counter    Rollout  CounterPlusRollout           Timestamp\n",
-      "0  0.000000  0.000000            0.000000 2025-01-01 00:00:00\n",
+      "0  37.270658 -59.474345          -22.203688 2025-01-01 00:00:00\n",
-      "1  0.547912  0.284967            0.832879 2025-01-01 00:00:01\n",
+      "1  43.561734 -54.184658          -10.622924 2025-01-01 00:00:01\n",
-      "2  0.425669  0.842959            1.268628 2025-01-01 00:00:02\n",
+      "2  42.158150 -43.826926           -1.668776 2025-01-01 00:00:02\n",
-      "3  1.142865  0.112064            1.254928 2025-01-01 00:00:03\n",
+      "3  50.392926 -57.394165           -7.001239 2025-01-01 00:00:03\n",
-      "4  1.537601  0.184200            1.721801 2025-01-01 00:00:04\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,
+   "execution_count": 1,
-   "id": "e7c8eeb1",
+   "id": "c61be7ab",
    "metadata": {},
    "outputs": [
     {
-     "ename": "TypeError",
+     "name": "stdout",
-     "evalue": "MLFlowRepository.__init__() missing 1 required positional argument: 'logger'",
+     "output_type": "stream",
-     "output_type": "error",
+     "text": [
-     "traceback": [
+      "[ 0.5479121  -0.12224312  0.71719584  0.39473606 -0.8116453   0.9512447\n",
-      "\u001b[31m---------------------------------------------------------------------------\u001b[39m",
+      "  0.5222794   0.57212861 -0.74377273 -0.09922812 -0.25840395  0.85352998\n",
-      "\u001b[31mTypeError\u001b[39m                                 Traceback (most recent call last)",
+      "  0.28773024  0.64552323 -0.1131716  -0.54552256  0.10916957 -0.87236549\n",
-      "\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",
+      "  0.65526234  0.2633288   0.51617548 -0.29094806  0.94139605  0.78624224\n",
-      "\u001b[31mTypeError\u001b[39m: MLFlowRepository.__init__() missing 1 required positional argument: 'logger'"
+      "  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",
+    "# Set random seed for reproducibility\n",
-    "    host=\"http://localhost:5080/\",\n",
+    "rng = np.random.default_rng(42)\n",
-    "    username=\"aignosi\",\n",
-    "    password=\"aignosi\"\n",
-    ")\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": {

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]},
+        'data': {
-        'timestamp': '2023-05-26 11:12:27',
+            '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],
+        'data': {
-                 'response_time': [0.1, 0.2, 0.3],
+            'prediction': {
-                 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
+                '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],
+        'data': {
-                 'response_time': [0.1, 0.2, 0.3],
+            'prediction': {
-                 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
+                '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'

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 pandas import DataFrame, Timestamp
-from pytest import fixture, mark
+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': [
+        'data': {
-            {'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0},
+            "variable": {
-            {'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0},
+                "2024-01-01": "var1",
-            {'timestamp': '2024-01-02', 'variable': 'var1', 'value': 3.0},
+                "2024-01-02": "var2",
-            {'timestamp': '2024-01-02', 'variable': 'var2', 'value': 4.0}
+                "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']
     )
 
 

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': {
+    assert output['content'] == {
-        0: 2,
+        'prediction': {
-        1: 3
+            'index_1': 2,
-    }, 'response_time': ANY}
+            '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,

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'],
+            'model_config': input_data['model_config'],
-            **metadata
         }, 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'],
+            'model_config': input_data['model_config'],
-            **metadata
         }, 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,
+            'model_config': model_config,
-            'opc_output_config': {'test': '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,
+            'model_config': model_config,
-            'opc_output_config': {'test': 'config'}
+            'opc_output_config': {'test': 'config'},
+            'prediction_store_policy': prediction_store_policy
         }, confidence, last_timestamp, ""
     )
 

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([

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"