Merge pull request #23 from Aignosi/SIENTIAPDE-1222-ajustar-a-library-para-fazer-o-download-do-courier
SIENTIAPDE-1222/SIENTIAPDE-1214: Refactor MLFlow, Enhance Logging, and Update Configuration Handling
This commit is contained in:
Matheus Demoner
4
.gitignore
vendored
4
.gitignore
vendored
@@ -43,4 +43,6 @@ git_key*
|
||||
|
||||
git_log
|
||||
|
||||
.env
|
||||
.env
|
||||
|
||||
tmp/
|
||||
196
README.md
196
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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
56
tests.ipynb
56
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": [
|
||||
"<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": {
|
||||
|
||||
@@ -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']
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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, ""
|
||||
)
|
||||
|
||||
|
||||
@@ -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([
|
||||
|
||||
10
values.yaml
10
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"
|
||||
|
||||
Reference in New Issue
Block a user