diff --git a/.gitignore b/.gitignore index b14c9ac..cfaf446 100644 --- a/.gitignore +++ b/.gitignore @@ -43,4 +43,6 @@ git_key* git_log -.env \ No newline at end of file +.env + +tmp/ \ No newline at end of file diff --git a/README.md b/README.md index b2c37be..500153c 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,7 @@ The Laborious system uses a Temporal-based workflow architecture with clear sepa - Health check endpoints for Kubernetes liveness/readiness probes - Graceful shutdown with cleanup procedures - Multi-instance deployment support + - Two dedicated task queues: `predictions_batch-queue` and `minimal_retrain-queue` #### **Workflows (`laborious/workflows/`)** - **PredictionsBatch**: Main entry point for batch prediction pipelines @@ -79,25 +80,32 @@ The Laborious system uses a Temporal-based workflow architecture with clear sepa - Configurable timeout and retry strategies #### **Activities (`laborious/activities/`)** +- **Activities**: Main activity orchestrator combining all functionality through multiple inheritance - **Gates**: Data quality validation and filtering mechanisms - **MLFlow**: Model transformation and prediction operations - **OPC**: Real-time data export to industrial OPC servers -- **Activities**: Main activity orchestrator and coordination - **Key Features**: + - Multiple inheritance pattern for unified activity interface - Configurable filter policies and validation rules - - MLFlow model serving integration + - MLFlow model serving integration with configurable flavors - OPC UA client with certificate-based authentication - - Comprehensive error handling and notification + - Comprehensive error handling and notification integration + - Support for multiple OPC servers with independent configurations #### **Data Services (`laborious/utils/`)** -- **Connectors**: Database and external service configuration management +- **Connectors Config**: Environment variable-based configuration management - **Repository**: Data access layer for MLFlow and OPC operations + - `model_repository.py`: MLFlow model operations and retraining + - `opc_repository.py`: OPC server communication and data writing - **Filters**: Data quality validation and MLFlow response filtering + - `conditional_filters.py`: Input data validation filters + - `mlflow_filters.py`: MLFlow API response validation filters - **Key Features**: - - Environment variable-based configuration + - Environment variable-based configuration with sensible defaults - Connection pool management and optimization - Security credential management - Configuration validation and error handling + - Support for multiple OPC servers and MLFlow model flavors ### Data Flow Architecture @@ -506,23 +514,32 @@ pytest tests/workflow/test_predictions_batch.py ## 📊 Monitoring and Metrics -The Laborious system exposes comprehensive Prometheus metrics: +The Laborious system exposes comprehensive Prometheus metrics for operational visibility and performance monitoring: -### Application Metrics +### Application Health Metrics - `app_up`: Application health status (1=healthy, 0=unhealthy) -- `laborious_predictions_written_count`: Prediction export operation count -- `laborious_prediction_confidence_monitor`: Prediction confidence monitoring -- `laborious_prediction_response_time_monitor`: Prediction response time monitoring + - Labels: `pod_id` -### MLFlow Metrics -- Model transformation and prediction success rates -- API response times and error rates -- Model retention and versioning metrics +### Prediction Operation Metrics +- `laborious_predictions_written_count`: Counter for successful prediction exports + - Labels: `pod_id`, `model_name`, `pipeline_name` +- `laborious_prediction_confidence_monitor`: Gauge for current prediction confidence levels + - Labels: `pod_id`, `model_name`, `pipeline_name` +- `laborious_prediction_response_time_monitor`: Histogram for prediction response times + - Labels: `pod_id`, `model_name`, `pipeline_name` + - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] -### Export Metrics -- PostgreSQL export operation counts and response times -- OPC server write operations and performance -- Data quality filter pass/fail rates +### OPC Export Metrics +- `laborious_prediction_opc_writing_count`: Counter for OPC server write operations + - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id` +- `laborious_prediction_opc_writing_response_time_monitor`: Histogram for OPC write response times + - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id` + - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] + +### Data Quality Metrics +- Filter pass/fail rates through notification system +- MLFlow API response validation metrics +- Data quality gate performance tracking ## ⚙️ Configuration @@ -537,14 +554,33 @@ The Laborious system exposes comprehensive Prometheus metrics: | `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes | | `POSTGRES_PASSWORD` | PostgreSQL password | `sientia` | Yes | | `POSTGRES_DBNAME` | PostgreSQL database | `sientia` | Yes | -| `MLFLOW_HOST` | MLFlow server hostname | `localhost` | Yes | -| `MLFLOW_PORT` | MLFlow server port | `5000` | Yes | -| `MLFLOW_USERNAME` | MLFlow username | `admin` | Yes | -| `MLFLOW_PASSWORD` | MLFlow password | `admin` | Yes | +| `POSTGRES_MIN_CONNECTIONS` | Minimum PostgreSQL connections | `5` | No | +| `POSTGRES_MAX_CONNECTIONS` | Maximum PostgreSQL connections | `20` | No | +| `MLFLOW_HOST` | MLFlow server hostname | `http://localhost` | Yes | +| `MLFLOW_PORT` | MLFlow server port | `5080` | Yes | +| `MLFLOW_USERNAME` | MLFlow username | `aignosi` | Yes | +| `MLFLOW_PASSWORD` | MLFlow password | `aignosi` | Yes | | `OPC_CONFIG` | OPC server configuration (JSON) | `{}` | No | -| `MONGODB_URL` | MongoDB connection URI | `localhost:27017` | Yes | +| `OPC_ID` | OPC server identifier | `1` | No | +| `OPC_URL` | OPC server URL | `opc.tcp://localhost:4840` | No | +| `OPC_SERVER_URI` | OPC server URI | `opc.tcp://localhost:4840` | No | +| `OPC_CERT_PATH` | OPC client certificate path | `None` | No | +| `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No | +| `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No | +| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No | +| `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes | +| `MONGODB_USERNAME` | MongoDB username | `root` | Yes | +| `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes | +| `MONGODB_DATABASE_NAME` | MongoDB database name | `sientia` | Yes | +| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index hours | `1` | No | +| `LOG_LEVEL` | Application log level | `INFO` | No | +| `PROJECT_NAME` | Project name for metrics | `laborious` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | +| `POD_ID` | Kubernetes pod identifier | `None` | No | + + + ### OPC Configuration @@ -586,74 +622,88 @@ For single OPC server, use individual environment variables: MongoDB pipeline configuration: -#### Predictions Batch Workflow +#### Predictions Batch Workflow configuration sample + +This is the configuration for the Predictions Batch Workflow, to be inserted into the MongoDB pipeline collection. + ```json { "schedule_name": "laborious-orchestrated-pipeline", "model_id": "1", "workflow_type": "predictions_batch", - "frequency": "30s", # Workflow execution frequency - "max_retry_policy": 1, # Maximum number of retries for the workflow + "frequency": "30s", + "max_retry_policy": 1, "query": "select * from sientia_data.laborious_data where model_id = 1 and \"timestamp\" > NOW() - INTERVAL '5 minutes' order by \"timestamp\" desc limit 30;", - "retention_time": 60, # Retention time for models in minutes "write_tags": [ { - "server_id": "1", - "type": "prediction", # Type of tag to write, can be prediction or confidence + "server_id": "server1", + "type": "prediction", "addr": "ns=2;i=5", "data_type": "double" }, { - "server_id": "1", + "server_id": "server1", "type": "confidence", - "addr": "ns=2;i=5", + "addr": "ns=2;i=6", "data_type": "double" } ], - "input_filters": [ - { - "filter_name": "EMPTY_DATA", # Required filter - "policy": "STOP" - }, - { - "filter_name": "SPECIFIC_VARIABLES_NULL_VALUES", - "policy": "CONTINUE", - "config": { - "variables": [ - "Counter" - ] - } + "input_filters": { + "EMPTY_DATA": {"POLICY": "STOP"}, + "SPECIFIC_VARIABLES_NULL_VALUES": { + "POLICY": "CONTINUE", + "config": {"variables": ["Counter"]} } - ], - "mlflow_transform_filters": [ - { - "filter_name": "API_ERROR", # Required filter - "policy": "REPEAT" - }, - { - "filter_name": "NAN_VALUES", - "policy": "STOP" - } - ], - "mlflow_predict_filters": [ - { - "filter_name": "API_ERROR", # Required filter - "policy": "CONTINUE" - } - ], - "path_priority": [ # In case of multiple filters catch problems, this will determine the path to take - "STOP", - "CONTINUE", - "REPEAT" - ], + }, + "mlflow_transform_filters": { + "API_ERROR": {"POLICY": "REPEAT"}, + "NAN_VALUES": {"POLICY": "STOP"} + }, + "mlflow_predict_filters": { + "API_ERROR": {"POLICY": "CONTINUE"} + }, + "path_priority": ["STOP", "CONTINUE", "REPEAT"], "active": true, - "datetime_columns": [ # Columns in data comming from query that are datetime - "timestamp", - "created_at" - ], "updated_at": { - "$date": "2025-08-27T18:35:01.600Z" - } + "$date": "2025-09-16T10:00:00.000Z" + }, + "datetime_columns": ["timestamp", "created_at"], + "predictions_storage_policy": "lts:1" +} +``` + +This is the configuration created by the Orchestrator in Temporal. + +```json +{ + "datetime_columns":["timestamp","created_at"], + "frequency":"15m", + "input_filters":{"EMPTY_DATA":{"config":{},"policy":"STOP"}}, + "max_retry_policy":1, + "mlflow_predict_filters":{"API_ERROR":{"config":{},"policy":"CONTINUE"}}, + "mlflow_transform_filters":{ + "API_ERROR":{"config":{},"policy":"CONTINUE"}, + "EMPTY_DATA":{"config":{},"policy":"STOP"} + }, + "model_config":{ + "is_compressed":true, + "predict_flavor":"pyfunc", + "retention_minutes":60, + "retention_target":"artifact", + "transform_function_keyword":"transform" + }, + "model_id":"352", + "model_name":"courier", + "opc_output_config":{}, + "path_priority":["STOP","CONTINUE","REPEAT"], + "predictions_storage_policy":"lts:1", + "query":"select * from sientia_data.laborious_data where model_id = 352 order by \"timestamp\" desc limit 300;", + "retention_time":3600, + "schedule_name":"laborious-courier", + "schema":"sientia_data", + "table_name":"predictions", + "updated_at":"2025-09-12 19:35:01.600000+0000", + "workflow_type":"predictions_batch" } ``` @@ -667,7 +717,7 @@ laborious/ │ ├── gates.py # Data quality gates and filtering │ ├── mlflow.py # MLFlow model operations │ └── opc.py # OPC server operations -├── workflow/ # Temporal workflow definitions +├── workflows/ # Temporal workflow definitions │ ├── predictions_batch.py # Main batch prediction workflow │ ├── minimal_retrain.py # Model retraining workflow │ └── sub_workflows/ # Sub-workflow implementations diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 34c26f1..4a71455 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -8,6 +8,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.base import BaseActivity from sientia_do.observability.logger import Logger from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now + from sientia_do.formatters import create_sample_dict from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter from typing import Any from laborious.utils.filters.conditional_filters import ( @@ -122,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 @@ -207,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 = [] @@ -292,8 +292,8 @@ class Gates(BaseActivity): filter_output = [] - self.debug(f"Input data:\n {data}", metadata) - self.debug(f"Filters: {filters}", metadata) + self.debug(f"Input data:\n {data.head(5).to_string()}", metadata) + self.debug(f"Filters: \n {create_sample_dict(filters)}", metadata) for fil, config in filters.items(): if fil not in mlflow_content_filter_functions: @@ -410,7 +410,7 @@ class Gates(BaseActivity): self.debug( f"Prediction store policy: {prediction_store_policy}", metadata) - self.debug(f"Prediction data: {data.to_string()}", 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) @@ -445,7 +445,7 @@ class Gates(BaseActivity): data = data.reset_index(drop=True) 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() @@ -521,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) diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 3e3549c..bd06283 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -1,12 +1,15 @@ -import json from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): + from datetime import datetime + from pandas import Timestamp, to_datetime + from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger + from sientia_do.formatters import create_sample_dict from laborious.utils.repository.model_repository import MLFlowRepository from typing import Any import numpy as np @@ -57,7 +60,7 @@ class MLFlow(BaseActivity): self.mlflow_password = mlflow_password self.model_monitoring_repository = MLFlowRepository( - f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password, logger.base_logger + f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password, logger ) @activity.defn(name="request_transform") @@ -94,10 +97,10 @@ class MLFlow(BaseActivity): self.info('Transforming data...', metadata) data = DataFrame(input_data['data']) model_name = input_data['model_name'] - model_retention = input_data['model_retention'] + model_config = input_data.get('model_config', {}) self.debug("Raw input data:", metadata) - self.debug(data, metadata) + self.debug(data.head(5).to_string(), metadata) # Sort by created_at in descending order and keep first occurrence of each variable/timestamp pair data = data.sort_values('created_at', ascending=False).drop_duplicates( @@ -113,14 +116,18 @@ class MLFlow(BaseActivity): data.columns.name = None self.debug("Processed input data:", metadata) - self.debug(data, metadata) + self.debug(data.head(5).to_string(), metadata) # Request transformation from MLFlow model response_data = self.model_monitoring_repository.transform( - model_name, data, model_retention) + model_name, data, model_config, metadata + ) - self.debug("Transform response data:", metadata) - self.debug(json.dumps(response_data, indent=4), metadata) + self.debug( + f"Transform raw response data: \n {create_sample_dict(response_data, max_items=5, max_depth=5)}", metadata) + + self.debug( + f"Transform response data: \n {create_sample_dict(response_data, max_items=5, max_depth=5)}", metadata) self.info("Data transformed successfully", metadata) @@ -160,19 +167,24 @@ class MLFlow(BaseActivity): self.info('Predicting data...', metadata) data = DataFrame(input_data['data']) model_name = input_data['model_name'] - model_retention = input_data['model_retention'] + model_config = input_data.get('model_config', {}) - self.debug(data, metadata) + self.debug(f"Input data for: \n {data.head(5).to_string()}", metadata) # Convert numpy.nan to None for model compatibility data.replace(np.nan, None, inplace=True) + data['timestamp'] = data.index + data['timestamp'] = to_datetime( + data['timestamp'], format=DATETIME_FORMAT_WITH_TZ).dt.strftime(DATETIME_FORMAT) + # Request prediction from MLFlow model response_data = self.model_monitoring_repository.predict( - model_name, data, model_retention) + model_name, data, model_config, metadata + ) - self.debug("Prediction response data:", metadata) - self.debug(json.dumps(response_data, indent=4), metadata) + self.debug( + f"Prediction response data: \n {create_sample_dict(response_data, max_items=5, max_depth=5)}", metadata) self.info("Data predicted successfully", metadata) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 68322f3..3107d1c 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -16,16 +16,57 @@ import pandas as pd import mlflow from os import makedirs, path, remove from sientia.ModelServing import ModelServing +from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ +from sientia_do.observability.logger import Logger class MLFlowRepository(): - def __init__(self, host, username, password, logger): + def __init__(self, host, username, password, logger: Logger): self.model_serving = ModelServing(tracking_uri=host, username=username, password=password, logger=logger) + self.logger = logger - def transform(self, model_name: str, data: pd.DataFrame, model_retention: int): + def detect_and_parse_datetime_index(self, data: pd.DataFrame, metadata: dict) -> pd.DataFrame: + """ + Detect and parse datetime index from data. index must be a timestamp like column. + This function must detect the timestamp type (pandas Timestamp or datetime) and convert it to DATETIME_FORMAT_WITH_TZ. + If the index is a string, must be in format DATETIME_FORMAT_WITH_TZ. + If another type or format, must raise an error. + """ + index = data.index + + # Get type of first element of index + index_type = type(index[0]) + + self.logger.custom_info(f"Index type: {index_type}", metadata) + + message = f"Index must be all timestamp like column. Valid formats are: pandas Timestamp, datetime, string in format {DATETIME_FORMAT_WITH_TZ}" + + # Check if all in index are of the same type + if not all(isinstance(i, index_type) for i in index): + raise ValueError( + f"{message}") + + # Check type and converts to DATETIME_FORMAT_WITH_TZ + if index_type == str: + # Validate format of string and return error if not valid + try: + pd.to_datetime(data.index, format=DATETIME_FORMAT_WITH_TZ) + except ValueError: + raise ValueError( + f"{message}") + + elif index_type == datetime or index_type == pd.Timestamp: + data.index = data.index.strftime(DATETIME_FORMAT_WITH_TZ) + else: + raise ValueError( + f"{message}") + + return data + + def transform(self, model_name: str, data: pd.DataFrame, model_config: dict, metadata: dict) -> dict: """ Transform data using a model. @@ -39,11 +80,30 @@ class MLFlowRepository(): """ try: + self.logger.custom_debug( + f"Data received for model transformation: {data.to_csv()}", metadata) + + model_retention = model_config.get('retention_minutes', 0) + flavor = model_config.get('transform_flavor', 'sklearn') + compressed = model_config.get('is_compressed', False) + retention_target = model_config.get('retention_target', 'model') + transform_keyword = model_config.get( + 'transform_function_keyword', 'predict') + + transformed_data = self.model_serving.get_cached_transform( + model_name, data, model_retention, flavor, + compressed, retention_target, transform_keyword + ) + + self.logger.custom_debug( + f"Data received from model transformation: {transformed_data.to_csv()}", metadata) + + transformed_data = self.detect_and_parse_datetime_index( + transformed_data, metadata) return { 'success': True, - 'content': self.model_serving.get_cached_transform( - model_name, data, model_retention).to_dict() + 'content': transformed_data.to_dict() } except Exception as e: @@ -55,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. @@ -68,14 +128,25 @@ 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() diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index de1fbbe..e522eaa 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -87,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 @@ -113,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 diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index ae3ccfb..8e7df07 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -79,8 +79,7 @@ class FormatAndExportPrediction(): 'timestamp': input_data['timestamp'], 'model_id': input_data['model_id'], 'prediction_confidence': prediction_confidence, - 'prediction_store_policy': input_data.get( - 'prediction_store_policy', 'lts:1') + 'prediction_store_policy': input_data['prediction_store_policy'] }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index e1a0162..777fa1c 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -79,7 +79,7 @@ class PredictionProcess(): data = input_data['data'] model_id = input_data['model_id'] model_name = input_data['model_name'] - model_retention = input_data['model_retention'] + model_config = input_data.get('model_config', {}) # Get last timestamp for incremental processing last_timestamp = await workflow.execute_local_activity_method( @@ -120,10 +120,10 @@ class PredictionProcess(): **metadata, 'data': data, 'model_name': model_name, - 'model_retention': model_retention + 'model_config': model_config }, retry_policy=retry_policy, - start_to_close_timeout=timedelta(minutes=1), + start_to_close_timeout=timedelta(minutes=5), ) # Validate MLFlow transform response @@ -172,10 +172,10 @@ class PredictionProcess(): **metadata, 'data': transformed_data, 'model_name': model_name, - 'model_retention': model_retention + 'model_config': model_config }, retry_policy=retry_policy, - start_to_close_timeout=timedelta(minutes=1), + start_to_close_timeout=timedelta(minutes=5), ) # Validate MLFlow prediction response @@ -209,11 +209,12 @@ class PredictionProcess(): 'timestamp': last_timestamp, 'model_id': model_id, 'model_name': model_name, - 'model_retention': model_retention, + 'model_config': model_config, 'opc_output_config': input_data['opc_output_config'], 'schema': input_data['schema'], 'table_name': input_data['table_name'], - 'comment': comment + 'comment': comment, + 'prediction_store_policy': input_data['prediction_store_policy'] } ) @@ -248,7 +249,7 @@ class PredictionProcess(): table_name = input_data['table_name'] model_id = input_data['model_id'] model_name = input_data['model_name'] - model_retention = input_data['model_retention'] + model_config = input_data.get('model_config', {}) path_flag = path_flag.upper() if path_flag else '' @@ -282,11 +283,12 @@ class PredictionProcess(): 'timestamp': last_timestamp, 'model_id': model_id, 'model_name': model_name, - 'model_retention': model_retention, + 'model_config': model_config, 'schema': schema, 'table_name': table_name, 'comment': comment, - 'opc_output_config': input_data['opc_output_config'] + 'opc_output_config': input_data['opc_output_config'], + 'prediction_store_policy': input_data['prediction_store_policy'] } ) return True diff --git a/requirements.txt b/requirements.txt index 3f723f7..1ce9f9c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,6 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.5 -git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.13 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6 +git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 prometheus-client diff --git a/tests.ipynb b/tests.ipynb index 74750bf..6ec555e 100644 --- a/tests.ipynb +++ b/tests.ipynb @@ -165,7 +165,7 @@ }, { "cell_type": "code", - "execution_count": 8, + "execution_count": 1, "id": "c61be7ab", "metadata": {}, "outputs": [ @@ -404,6 +404,60 @@ "print(len(b))\n", "print(b.size)\n" ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f3374174", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "\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": { diff --git a/tests/laborious/activities/test_mlflow.py b/tests/laborious/activities/test_mlflow.py index 7aced7c..85cb7f1 100644 --- a/tests/laborious/activities/test_mlflow.py +++ b/tests/laborious/activities/test_mlflow.py @@ -1,8 +1,10 @@ +from datetime import datetime from unittest.mock import ANY, MagicMock, patch import numpy as np -from pandas import DataFrame -from pytest import fixture, mark +from pandas import DataFrame, Timestamp +from pytest import fixture, mark, raises +from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ from laborious.activities.mlflow import MLFlow from sientia_do.notifications.models import NotificationLevel @@ -58,7 +60,7 @@ metadata = { @mark.asyncio @patch("laborious.activities.mlflow.DataFrame") @patch("laborious.activities.mlflow.max") -async def test_request_transform(mock_max, mock_dataframe, mlflow): +async def test_request_transform_success(mock_max, mock_dataframe, mlflow): mock_max.return_value = '2024-01-02' # Mock input data input_data = { @@ -78,7 +80,7 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow): 'value': 1.0, 'created_at': '2024-01-01 12:00:00'} ], 'model_name': 'test_model', - 'model_retention': 30 + 'model_config': {} } # Mock the transform response @@ -107,26 +109,35 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow): # Verify the repository was called with correct arguments mlflow.model_monitoring_repository.transform.assert_called_once_with( - 'test_model', mock_dataframe, 30 + 'test_model', mock_dataframe, {}, metadata['metadata'] ) @mark.asyncio @patch("laborious.activities.mlflow.DataFrame") +@patch("laborious.activities.mlflow.to_datetime") @patch("laborious.activities.mlflow.max") -async def test_request_predict(mock_max, mock_dataframe, mlflow): +async def test_request_predict(mock_max, mock_to_datetime, mock_dataframe, mlflow): mock_max.return_value = '2024-01-02' # Mock input data input_data = { **metadata, - 'data': [ - {'timestamp': '2024-01-01', 'variable': 'var1', 'value': 1.0}, - {'timestamp': '2024-01-01', 'variable': 'var2', 'value': 2.0}, - {'timestamp': '2024-01-02', 'variable': 'var1', 'value': 3.0}, - {'timestamp': '2024-01-02', 'variable': 'var2', 'value': 4.0} - ], + 'data': { + "variable": { + "2024-01-01": "var1", + "2024-01-02": "var2", + "2024-01-03": "var1", + "2024-01-04": "var2" + }, + "value": { + "2024-01-01": 1.0, + "2024-01-02": 2.0, + "2024-01-03": 3.0, + "2024-01-04": 4.0 + } + }, 'model_name': 'test_model', - 'model_retention': 30 + 'model_config': {} } # Mock the predict response @@ -140,13 +151,23 @@ async def test_request_predict(mock_max, mock_dataframe, mlflow): mock_dataframe.return_value.replace.assert_called_once_with( np.nan, None, inplace=True ) + mock_dataframe.return_value.__setitem__.assert_any_call( + 'timestamp', mock_to_datetime.return_value.dt.strftime.return_value + ) + + mock_to_datetime.assert_called_once_with( + mock_dataframe.return_value.__getitem__.return_value, format=DATETIME_FORMAT_WITH_TZ + ) + mock_to_datetime.return_value.dt.strftime.assert_called_once_with( + DATETIME_FORMAT + ) # Verify the response assert response_data == expected_response # Verify the repository was called with correct arguments mlflow.model_monitoring_repository.predict.assert_called_once_with( - 'test_model', mock_dataframe.return_value, 30 + 'test_model', mock_dataframe.return_value, {}, metadata['metadata'] ) diff --git a/tests/laborious/utils/repository/test_model_repository.py b/tests/laborious/utils/repository/test_model_repository.py index 8b3716e..cdc59b4 100644 --- a/tests/laborious/utils/repository/test_model_repository.py +++ b/tests/laborious/utils/repository/test_model_repository.py @@ -2,7 +2,8 @@ from unittest.mock import ANY, MagicMock, call, patch import numpy as np from pandas import DataFrame import pytest -from laborious.utils.repository import model_repository +from datetime import datetime, timezone +from pandas import Timestamp from laborious.utils.repository.model_repository import MLFlowRepository @@ -22,32 +23,132 @@ def mlflow_repository(): return repo +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + + +class Any: + pass + + +invalid_cases = [ + ( + { + 'value': { + '2024-01-01 12:00:00': 1, + 2024: 2 + } + } + ), + ( + { + 'value': { + '2024-01-01': 1, + '2024-01-02': 2 + } + } + ), + ( + { + 'value': { + Any(): 1, + Any(): 2 + } + } + ) +] + + +@pytest.mark.parametrize("data", invalid_cases) +def test_detect_and_parse_datetime_index_error_cases(mlflow_repository, data): + input_data = DataFrame( + data + ) + + with pytest.raises(ValueError) as e: + mlflow_repository.detect_and_parse_datetime_index( + input_data, metadata['metadata']) + + assert str(e) == "Index must be all timestamp like column. Valid formats are: pandas Timestamp, datetime, string in format %Y-%m-%d %H:%M:%S" + + +valid_cases = [ + ( + { + 'value': { + '2024-01-01 12:00:00+0000': 1, + '2024-01-02 12:00:00+0000': 2 + } + }, ['2024-01-01 12:00:00+0000', '2024-01-02 12:00:00+0000'] + ), + ( + { + 'value': { + datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc): 1, + datetime(2025, 1, 2, 12, 0, 0, tzinfo=timezone.utc): 2 + } + }, ['2025-01-01 12:00:00+0000', '2025-01-02 12:00:00+0000'] + ), + ( + { + 'value': { + Timestamp(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc): 1, + Timestamp(2026, 1, 2, 12, 0, 0, tzinfo=timezone.utc): 2 + } + }, ['2026-01-01 12:00:00+0000', '2026-01-02 12:00:00+0000'] + ), +] + + +@pytest.mark.parametrize("data,expected", valid_cases) +def test_detect_and_parse_datetime_index_valid_format(mlflow_repository, data, expected): + input_data = DataFrame(data) + + response = mlflow_repository.detect_and_parse_datetime_index( + input_data, metadata['metadata']) + + assert response.index.tolist() == expected + + def test_transform_success(mlflow_repository): - data = 'data' + data = MagicMock() model_name = 'model' - output = mlflow_repository.transform(model_name, data, 1) + mlflow_repository.detect_and_parse_datetime_index = MagicMock() + + output = mlflow_repository.transform( + model_name, data, {}, metadata['metadata']) mlflow_repository.model_serving.get_cached_transform.assert_called_once_with( - model_name, data, 1) + model_name, data, 0, 'sklearn', False, 'model', 'predict') + + mlflow_repository.detect_and_parse_datetime_index.assert_called_once_with( + mlflow_repository.model_serving.get_cached_transform.return_value, metadata['metadata']) assert output == { 'success': True, - 'content': mlflow_repository.model_serving.get_cached_transform.return_value.to_dict.return_value + 'content': mlflow_repository.detect_and_parse_datetime_index.return_value.to_dict.return_value } def test_transform_error(mlflow_repository): - data = 'data' + data = MagicMock() model_name = 'model' mlflow_repository.model_serving.get_cached_transform.side_effect = Exception( 'error') - output = mlflow_repository.transform(model_name, data, 1) + output = mlflow_repository.transform( + model_name, data, {}, metadata['metadata']) mlflow_repository.model_serving.get_cached_transform.assert_called_once_with( - model_name, data, 1) + model_name, data, 0, 'sklearn', False, 'model', 'predict') assert output == { 'success': False, @@ -70,10 +171,11 @@ def test_predict_success(mlflow_repository): [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'] == { @@ -88,17 +190,23 @@ def test_predict_success(mlflow_repository): 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, diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index e76b586..df60ada 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -34,9 +34,12 @@ async def test_run(workflow_mock, prediction_process): 'mlflow_transform_filters': {'test': 'filter'}, 'mlflow_predict_filters': {'test': 'filter'}, 'model_name': 'test_model_name', - 'model_retention': '30', + 'model_config': { + 'retention': '30' + }, 'path_priority': ['continue', 'repeat', 'stop'], 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': 'lts:1' } # Mock the activity responses @@ -61,54 +64,54 @@ async def test_run(workflow_mock, prediction_process): workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.get_last_timestamp, { + **metadata, 'data': input_data['data'], - **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.input_gate, { + **metadata, 'filters': input_data['input_filters'], 'data': input_data['data'], 'path_priority': input_data['path_priority'], - **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { + **metadata, 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'], - **metadata + 'model_config': input_data['model_config'], }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_response_gate, { + **metadata, 'filters': input_data['mlflow_transform_filters'], 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'type': 'transform', 'path_priority': input_data['path_priority'], - **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { + **metadata, 'filters': input_data['mlflow_transform_filters'], 'data': 'transformed_data', 'type': 'transform', 'path_priority': input_data['path_priority'], - **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_predict, { + **metadata, 'data': 'transformed_data', 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'], - **metadata + 'model_config': input_data['model_config'], }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_response_gate, { + **metadata, 'filters': input_data['mlflow_predict_filters'], 'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'}, 'type': 'predict', 'path_priority': input_data['path_priority'], - **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_child_workflow.assert_called_once_with( @@ -121,11 +124,12 @@ async def test_run(workflow_mock, prediction_process): 'timestamp': '2024-01-01', 'model_id': 1, 'model_name': 'test_model_name', - 'model_retention': '30', + 'model_config': input_data['model_config'], 'opc_output_config': input_data['opc_output_config'], 'schema': input_data['schema'], 'table_name': input_data['table_name'], - 'comment': 'Error' + 'comment': 'Error', + 'prediction_store_policy': input_data['prediction_store_policy'] } ) @@ -145,7 +149,9 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process): 'mlflow_transform_filters': {'test': 'filter'}, 'mlflow_predict_filters': {'test': 'filter'}, 'model_name': 'test_model_name', - 'model_retention': '30', + 'model_config': { + 'retention': '30' + }, 'path_priority': ['continue', 'repeat', 'stop'], 'opc_output_config': {'test': 'config'} } @@ -164,13 +170,13 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process): workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.get_last_timestamp, { 'data': input_data['data'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY), call(Activities.input_gate, { 'filters': input_data['input_filters'], 'data': input_data['data'], 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY) ]) workflow_mock.execute_child_workflow.assert_not_called() @@ -191,7 +197,9 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ 'mlflow_transform_filters': {'test': 'filter'}, 'mlflow_predict_filters': {'test': 'filter'}, 'model_name': 'test_model_name', - 'model_retention': '30', + 'model_config': { + 'retention': '30' + }, 'path_priority': ['continue', 'repeat', 'stop'], 'opc_output_config': {'test': 'config'} } @@ -212,7 +220,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.get_last_timestamp, { 'data': input_data['data'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ @@ -220,14 +228,14 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ 'filters': input_data['input_filters'], 'data': input_data['data'], 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'], + 'model_config': input_data['model_config'], **metadata }, retry_policy=ANY, start_to_close_timeout=ANY) @@ -260,7 +268,9 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process 'mlflow_transform_filters': {'test': 'filter'}, 'mlflow_predict_filters': {'test': 'filter'}, 'model_name': 'test_model_name', - 'model_retention': '30', + 'model_config': { + 'retention': '30' + }, 'path_priority': ['continue', 'repeat', 'stop'], 'opc_output_config': {'test': 'config'} } @@ -285,7 +295,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.get_last_timestamp, { 'data': input_data['data'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ @@ -293,14 +303,14 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process 'filters': input_data['input_filters'], 'data': input_data['data'], 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'], + 'model_config': input_data['model_config'], **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ @@ -309,7 +319,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'type': 'transform', 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { @@ -317,7 +327,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process 'data': 'transformed_data', 'type': 'transform', 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_child_workflow.assert_not_called() @@ -338,7 +348,9 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p 'mlflow_transform_filters': {'test': 'filter'}, 'mlflow_predict_filters': {'test': 'filter'}, 'model_name': 'test_model_name', - 'model_retention': '30', + 'model_config': { + 'retention': '30' + }, 'path_priority': ['continue', 'repeat', 'stop'], 'opc_output_config': {'test': 'config'} } @@ -364,7 +376,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.get_last_timestamp, { 'data': input_data['data'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ @@ -372,14 +384,14 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p 'filters': input_data['input_filters'], 'data': input_data['data'], 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_transform, { 'data': input_data['data'], 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'], + 'model_config': input_data['model_config'], **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ @@ -388,7 +400,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p 'data': {'content': 'transformed_data', 'timestamp': '2024-01-01'}, 'type': 'transform', 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.mlflow_content_gate, { @@ -396,13 +408,13 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p 'data': 'transformed_data', 'type': 'transform', 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ call(Activities.request_predict, { 'data': 'transformed_data', 'model_name': input_data['model_name'], - 'model_retention': input_data['model_retention'], + 'model_config': input_data['model_config'], **metadata }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_local_activity_method.assert_has_calls([ @@ -411,7 +423,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p 'data': {'content': 'predicted_data', 'timestamp': '2024-01-01'}, 'type': 'predict', 'path_priority': input_data['path_priority'], - **metadata + **metadata, }, retry_policy=ANY, start_to_close_timeout=ANY)]) workflow_mock.execute_child_workflow.assert_not_called() @@ -428,7 +440,9 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process): model = 'test_model' last_timestamp = '2024-01-01' model_name = 'test_model_name' - model_retention = '30' + model_config = { + 'retention': '30' + } # Act result = await prediction_process.path_flag_handler( @@ -439,7 +453,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process): 'model_id': model, 'last_timestamp': last_timestamp, 'model_name': model_name, - 'model_retention': model_retention + 'model_config': model_config }, confidence, last_timestamp, "" ) @@ -461,7 +475,9 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): model = 'test_model' last_timestamp = '2024-01-01' model_name = 'test_model_name' - model_retention = '30' + model_config = { + 'retention': '30' + } # Act result = await prediction_process.path_flag_handler( @@ -472,7 +488,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): 'model_id': model, 'last_timestamp': last_timestamp, 'model_name': model_name, - 'model_retention': model_retention + 'model_config': model_config }, confidence, last_timestamp, "" ) @@ -505,7 +521,10 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): model = 'test_model' last_timestamp = '2024-01-01' model_name = 'test_model_name' - model_retention = '30' + model_config = { + 'retention': '30' + } + prediction_store_policy = 'erl:1' # Act result = await prediction_process.path_flag_handler( @@ -516,8 +535,9 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): 'model_id': model, 'last_timestamp': last_timestamp, 'model_name': model_name, - 'model_retention': model_retention, - 'opc_output_config': {'test': 'config'} + 'model_config': model_config, + 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy }, confidence, last_timestamp, 'Prediction Process' ) @@ -534,11 +554,12 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): 'timestamp': last_timestamp, 'model_id': model, 'model_name': model_name, - 'model_retention': model_retention, + 'model_config': model_config, 'schema': schema, 'table_name': table_name, 'comment': 'Prediction Process', - 'opc_output_config': {'test': 'config'} + 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy } ) @@ -555,8 +576,10 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process): model = 'test_model' last_timestamp = '2024-01-01' model_name = 'test_model_name' - model_retention = '30' - + model_config = { + 'retention': '30' + } + prediction_store_policy = 'erl:1' # Act result = await prediction_process.path_flag_handler( data, path_flag, { @@ -566,8 +589,9 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process): 'model_id': model, 'last_timestamp': last_timestamp, 'model_name': model_name, - 'model_retention': model_retention, - 'opc_output_config': {'test': 'config'} + 'model_config': model_config, + 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy }, confidence, last_timestamp, "" ) diff --git a/tests/laborious/workflows/test_predictions_batch.py b/tests/laborious/workflows/test_predictions_batch.py index e9b9bb6..90d7d21 100644 --- a/tests/laborious/workflows/test_predictions_batch.py +++ b/tests/laborious/workflows/test_predictions_batch.py @@ -33,7 +33,11 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch 'schema': 'test_schema', 'table_name': 'test_table', 'opc_output_config': 'test_opc_output_config', - 'datetime_columns': ['timestamp', 'created_at'] + 'datetime_columns': ['timestamp', 'created_at'], + 'prediction_store_policy': 'erl:1', + 'model_config': { + 'retention': '30' + } } await predictions_batch.run(input_data) @@ -72,9 +76,10 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch 'POLICY': 'STOP' } }), - 'model_retention': input_data.get('model_retention', 60), + 'model_config': input_data.get('model_config', {}), 'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']), - 'opc_output_config': input_data.get('opc_output_config', {}) + 'opc_output_config': input_data.get('opc_output_config', {}), + 'prediction_store_policy': input_data.get('prediction_store_policy', 'erl:1') } workflow_mock.execute_child_workflow.assert_has_calls([ diff --git a/values.yaml b/values.yaml index 716254a..4eb83d7 100644 --- a/values.yaml +++ b/values.yaml @@ -7,13 +7,13 @@ replicaCount: 1 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/ image: - repository: aignosi.azurecr.io/sientia-module + repository: aignosi.azurecr.io/sientia-module-courier # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.5" + 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-1182-ajustar-laborious-para-pegar-timestamp-da-resposta-do-mlflow" + 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"