Merge pull request #19 from Aignosi/SIENTIAPDE-1084-ajustar-documentacao
SIENTIAPDE-1084: Refactor OPC activities, update prediction workflow, and enhance documentation
This commit is contained in:
Bruno Domingues
29
.env.example
Normal file
29
.env.example
Normal file
@@ -0,0 +1,29 @@
|
||||
POSTGRES_HOST="paradedb-rw.paradedb.svc.cluster.local"
|
||||
POSTGRES_PORT="5432"
|
||||
POSTGRES_USER="sientia"
|
||||
POSTGRES_PASSWORD="password"
|
||||
POSTGRES_DBNAME="sientia"
|
||||
POSTGRES_MIN_CONNECTIONS="10"
|
||||
POSTGRES_MAX_CONNECTIONS="30"
|
||||
|
||||
MLFLOW_HOST="http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local"
|
||||
MLFLOW_PORT="80"
|
||||
MLFLOW_USERNAME="aignosi"
|
||||
MLFLOW_PASSWORD="mlflow_password"
|
||||
|
||||
OPC_ID="1"
|
||||
OPC_URL="opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840"
|
||||
|
||||
LOG_LEVEL="DEBUG"
|
||||
HTTP_METRICS_PORT="9090"
|
||||
HTTP_SDK_METRICS_PORT="9091"
|
||||
PROJECT_NAME="sientia-laborious"
|
||||
|
||||
TEMPORAL_HOST="temporal-frontend.temporal.svc.cluster.local:7233"
|
||||
TEMPORAL_NAMESPACE="laborious"
|
||||
|
||||
MONGODB_USERNAME="mongo_user"
|
||||
MONGODB_PASSWORD="mongo_db_password"
|
||||
MONGODB_URL="my-release-mongodb.mongodb.svc.cluster.local:27017"
|
||||
MONGODB_DATABASE="sientia"
|
||||
MONGODB_TTL_INDEX_HOURS="1"
|
||||
83
Dockerfile
83
Dockerfile
@@ -1,83 +0,0 @@
|
||||
FROM python:3.11-bookworm
|
||||
LABEL description="Deploy Mage on ECS"
|
||||
ARG FEATURE_BRANCH
|
||||
USER root
|
||||
SHELL ["/bin/bash", "-o", "pipefail", "-c"]
|
||||
|
||||
# Definir Python 3.11 como padrão
|
||||
ENV PATH="/usr/local/bin/python3.11:$PATH"
|
||||
RUN update-alternatives --install /usr/bin/python python /usr/local/bin/python3.11 1 && \
|
||||
update-alternatives --install /usr/bin/python3 python3 /usr/local/bin/python3.11 1 && \
|
||||
update-alternatives --config python3 <<< '1' && \
|
||||
update-alternatives --config python <<< '1'
|
||||
|
||||
## System Packages
|
||||
RUN \
|
||||
curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \
|
||||
curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \
|
||||
apt-get -y update && \
|
||||
ACCEPT_EULA=Y apt-get -y install --no-install-recommends \
|
||||
# NFS dependencies
|
||||
nfs-common \
|
||||
# odbc dependencies
|
||||
msodbcsql18 \
|
||||
unixodbc-dev \
|
||||
graphviz \
|
||||
# postgres dependencies
|
||||
postgresql-client \
|
||||
# R
|
||||
r-base && \
|
||||
apt-get clean && \
|
||||
rm -rf /var/lib/apt/lists/*
|
||||
|
||||
## R Packages
|
||||
RUN \
|
||||
R -e "install.packages('pacman', repos='http://cran.us.r-project.org')" && \
|
||||
R -e "install.packages('renv', repos='http://cran.us.r-project.org')"
|
||||
|
||||
## Python Packages
|
||||
RUN \
|
||||
pip3 install --no-cache-dir sparkmagic && \
|
||||
mkdir ~/.sparkmagic && \
|
||||
curl https://raw.githubusercontent.com/jupyter-incubator/sparkmagic/master/sparkmagic/example_config.json > ~/.sparkmagic/config.json && \
|
||||
sed -i 's/localhost:8998/host.docker.internal:9999/g' ~/.sparkmagic/config.json && \
|
||||
jupyter-kernelspec install --user "$(pip3 show sparkmagic | grep Location | cut -d' ' -f2)/sparkmagic/kernels/pysparkkernel"
|
||||
|
||||
# Mage integrations and other related packages
|
||||
RUN \
|
||||
pip3 install --no-cache-dir "git+https://github.com/wbond/oscrypto.git@d5f3437ed24257895ae1edd9e503cfb352e635a8" && \
|
||||
pip3 install --no-cache-dir "git+https://github.com/dremio-hub/arrow-flight-client-examples.git#egg=dremio-flight&subdirectory=python/dremio-flight" && \
|
||||
pip3 install --no-cache-dir "git+https://github.com/mage-ai/singer-python.git#egg=singer-python" && \
|
||||
pip3 install --no-cache-dir "git+https://github.com/mage-ai/dbt-mysql.git#egg=dbt-mysql" && \
|
||||
pip3 install --no-cache-dir "git+https://github.com/mage-ai/sqlglot#egg=sqlglot" && \
|
||||
pip3 install --no-cache-dir faster-fifo && \
|
||||
if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ]; then \
|
||||
pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git#egg=mage-integrations&subdirectory=mage_integrations"; \
|
||||
else \
|
||||
pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-integrations&subdirectory=mage_integrations"; \
|
||||
fi
|
||||
|
||||
# Mage
|
||||
COPY ./mage_ai/server/constants.py /tmp/constants.py
|
||||
RUN if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ] ; then \
|
||||
tag=$(tail -n 1 /tmp/constants.py) && \
|
||||
VERSION=$(echo "$tag" | tr -d "'") && \
|
||||
pip3 install --no-cache-dir "mage-ai[all]==$VERSION"; \
|
||||
else \
|
||||
pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-ai[all]"; \
|
||||
fi
|
||||
|
||||
## Startup Script
|
||||
COPY --chmod=0755 ./scripts/install_other_dependencies.py ./scripts/run_app.sh /app/
|
||||
ENV MAGE_DATA_DIR="/home/src/mage_data"
|
||||
ENV PYTHONPATH="${PYTHONPATH}:/home/src"
|
||||
WORKDIR /home/src
|
||||
EXPOSE 6789
|
||||
EXPOSE 7789
|
||||
|
||||
# Copia o arquivo requirements.txt para o contêiner
|
||||
COPY requirements.txt /app/requirements.txt
|
||||
RUN pip3 install --no-cache-dir -r /app/requirements.txt
|
||||
|
||||
|
||||
CMD ["/bin/sh", "-c", "/app/run_app.sh"]
|
||||
822
README.md
822
README.md
@@ -1,80 +1,579 @@
|
||||
# Sientia DataOps Laborious
|
||||
|
||||
The Sientia DataOps Laborious is a Temporal-based workflow application that handles batch predictions and data processing for industrial data. It integrates with MLFlow for model management, PostgreSQL for data storage, and OPC for real-time data output. The module is designed to process data in a reliable and scalable manner using Temporal.io's workflow orchestration capabilities. It's get data from Scouter sinks, process it, make predictions using MLFlow models and generates metrics for the predictions.
|
||||
A high-performance, scalable machine learning prediction system built on Temporal.io for industrial data processing and ML model inference. The Laborious system provides enterprise-grade ML model management, batch prediction processing, and real-time data export capabilities with comprehensive data quality validation and monitoring.
|
||||
|
||||
## Key Features
|
||||
## Features
|
||||
|
||||
- Batch predictions using MLFlow models
|
||||
- Data transformation and preprocessing
|
||||
- Workflow orchestration using Temporal.io
|
||||
- Integration with PostgreSQL for data storage
|
||||
- OPC integration for real-time data output
|
||||
- Comprehensive error handling and notifications
|
||||
- Configurable data filters and quality gates
|
||||
- Scalable deployment architecture
|
||||
### Core Functionality
|
||||
- **Batch Prediction Processing**: High-throughput ML model inference using MLFlow models
|
||||
- **Temporal Workflow Orchestration**: Robust workflow management with automatic retry policies and fault tolerance
|
||||
- **Data Quality Gates**: Configurable filtering for data validation, MLFlow API responses, and custom validation rules
|
||||
- **Multi-Model Support**: Flexible ML model management with retention policies and versioning
|
||||
- **Real-time Data Export**: PostgreSQL persistence and OPC server integration for industrial systems
|
||||
- **Comprehensive Monitoring**: Prometheus metrics and detailed logging for operational visibility
|
||||
|
||||
## Workflows
|
||||
### Advanced Capabilities
|
||||
- **Incremental Data Processing**: Timestamp-based data loading to avoid reprocessing
|
||||
- **Configurable Data Retention**: Model retention policies with automatic cleanup
|
||||
- **Notification System**: Integrated alerting and notification management via MongoDB
|
||||
- **Scalable Architecture**: Kubernetes-ready deployment with horizontal scaling support
|
||||
- **Model Retraining**: Automated model retraining workflows with production model updates
|
||||
|
||||
### Predictions Batch
|
||||
The main workflow that orchestrates batch predictions. Steps:
|
||||
## Architecture
|
||||
|
||||
- prepare_activity: Prepares the activity with schedule and model information
|
||||
- load_custom_query: Loads data using a custom query
|
||||
- prediction_process: Executes the prediction process using the Prediction Process sub-workflow
|
||||
|
||||
#### Workflow inputs:
|
||||
|
||||
- `schedule_name`: The schedule name of the activity
|
||||
- `model_name`: The model name of the activity
|
||||
- `model_id`: The model id of the activity
|
||||
- `query`: The custom query to load data
|
||||
- `schema`: The schema of the data
|
||||
- `table_name`: The name of the table to process
|
||||
- `input_filters`: The filters to be applied during prediction
|
||||
- `mlflow_transform_filters`: The filters to be applied during prediction
|
||||
- `mlflow_predict_filters`: The filters to be applied during prediction
|
||||
- `model_retention`: The model retention period in minutes
|
||||
- `path_priority`: The path priority
|
||||
The Laborious system uses a Temporal-based workflow architecture with clear separation of concerns and robust error handling. The architecture is designed for high availability, scalability, and operational excellence in production ML environments.
|
||||
|
||||
|
||||
### Prediction Process
|
||||
Sub-workflow that handles individual prediction processing:
|
||||
### Architecture Principles
|
||||
|
||||
- get_last_timestamp: Gets the last timestamp of the data
|
||||
- input_gate: Filters input data based on configured rules
|
||||
- repeat_last_prediction: Repeats the last prediction if the data is empty
|
||||
- request_transform: Makes predictions using MLFlow models
|
||||
- mlflow_response_gate: Handles prediction or transform responses and filters
|
||||
- mlflow_content_gate: Filters transform responses based on configured rules
|
||||
- request_predict: Makes predictions using MLFlow models
|
||||
- format_and_export_prediction: Formats and exports predictions using the
|
||||
Format and Export Prediction sub-workflow
|
||||
#### 1. **Separation of Concerns**
|
||||
- **Worker Layer**: Manages Temporal workers, task queues, and application lifecycle
|
||||
- **Workflow Layer**: Orchestrates business logic and process coordination
|
||||
- **Activity Layer**: Implements specific operations and external system interactions
|
||||
- **Data Layer**: Handles data persistence, caching, and external service connections
|
||||
|
||||
### Format and Export Prediction
|
||||
Sub-workflow that handles prediction formatting and export:
|
||||
#### 2. **Fault Tolerance & Resilience**
|
||||
- **Automatic Retry Policies**: Configurable retry strategies for transient failures
|
||||
- **Circuit Breaker Pattern**: Prevents cascading failures in external service calls
|
||||
- **Graceful Degradation**: System continues operating with reduced functionality
|
||||
- **Comprehensive Error Handling**: Detailed error reporting and notification integration
|
||||
|
||||
- format_prediction: Formats prediction data if path flag is None
|
||||
- format_default_prediction: Formats default prediction data if path flag is not None
|
||||
- export_to_postgres: Exports formatted predictions to PostgreSQL
|
||||
- write_to_opc: Writes predictions to OPC server
|
||||
#### 3. **Scalability & Performance**
|
||||
- **Horizontal Scaling**: Multiple worker instances for load distribution
|
||||
- **Task Queue Isolation**: Separate queues for different workflow types
|
||||
- **Connection Pooling**: Optimized database and external service connections
|
||||
- **Asynchronous Processing**: Non-blocking operations for improved throughput
|
||||
|
||||
## Environment variables
|
||||
#### 4. **Observability & Monitoring**
|
||||
- **Prometheus Metrics**: Comprehensive system and business metrics
|
||||
- **Structured Logging**: Consistent log format with correlation IDs
|
||||
- **Health Checks**: Endpoint health monitoring and alerting
|
||||
- **Performance Tracing**: Request flow tracking and bottleneck identification
|
||||
|
||||
- `POSTGRES_HOST`
|
||||
- `POSTGRES_PORT`
|
||||
- `POSTGRES_USER`
|
||||
- `POSTGRES_PASSWORD`
|
||||
- `POSTGRES_DBNAME`
|
||||
- `POSTGRES_MIN_CONNECTIONS`
|
||||
- `POSTGRES_MAX_CONNECTIONS`
|
||||
### Key Components
|
||||
|
||||
- `MLFLOW_HOST`
|
||||
- `MLFLOW_PORT`
|
||||
- `MLFLOW_USERNAME`
|
||||
- `MLFLOW_PASSWORD`
|
||||
#### **Worker (`laborious/worker/worker.py`)**
|
||||
- **Purpose**: Main application orchestrator managing Temporal workers and task queues
|
||||
- **Responsibilities**:
|
||||
- Temporal client initialization and connection management
|
||||
- Worker lifecycle management and graceful shutdown
|
||||
- Task queue configuration and load balancing
|
||||
- Prometheus metrics server initialization
|
||||
- Notification handler setup and configuration
|
||||
- OPC server connection management
|
||||
- **Key Features**:
|
||||
- Automatic scaling with `PollerBehaviorAutoscaling`
|
||||
- Health check endpoints for Kubernetes liveness/readiness probes
|
||||
- Graceful shutdown with cleanup procedures
|
||||
- Multi-instance deployment support
|
||||
|
||||
- `OPC_CONFIG` - json string containing the opc configuration for multiple opc servers
|
||||
For single opc server use:
|
||||
#### **Workflows (`laborious/workflows/`)**
|
||||
- **PredictionsBatch**: Main entry point for batch prediction pipelines
|
||||
- **PredictionProcess**: Core prediction pipeline with MLFlow integration
|
||||
- **FormatAndExportPrediction**: Data formatting and export operations
|
||||
- **MinimalRetrain**: Automated model retraining and deployment
|
||||
- **Key Features**:
|
||||
- Temporal workflow definitions with retry policies
|
||||
- Child workflow orchestration and delegation
|
||||
- Comprehensive error handling and recovery
|
||||
- Configurable timeout and retry strategies
|
||||
|
||||
#### **Activities (`laborious/activities/`)**
|
||||
- **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**:
|
||||
- Configurable filter policies and validation rules
|
||||
- MLFlow model serving integration
|
||||
- OPC UA client with certificate-based authentication
|
||||
- Comprehensive error handling and notification
|
||||
|
||||
#### **Data Services (`laborious/utils/`)**
|
||||
- **Connectors**: Database and external service configuration management
|
||||
- **Repository**: Data access layer for MLFlow and OPC operations
|
||||
- **Filters**: Data quality validation and MLFlow response filtering
|
||||
- **Key Features**:
|
||||
- Environment variable-based configuration
|
||||
- Connection pool management and optimization
|
||||
- Security credential management
|
||||
- Configuration validation and error handling
|
||||
|
||||
### Data Flow Architecture
|
||||
|
||||
#### **1. Batch Prediction Pipeline**
|
||||
```
|
||||
Input Data (PostgreSQL) → Data Quality Gates → MLFlow Transform →
|
||||
MLFlow Prediction → Response Validation → Export (PostgreSQL + OPC)
|
||||
```
|
||||
|
||||
#### **2. Model Retraining Pipeline**
|
||||
```
|
||||
Training Data → Model Retraining → Quality Validation →
|
||||
Production Update → Notification & Monitoring
|
||||
```
|
||||
|
||||
#### **3. Real-time Export Pipeline**
|
||||
```
|
||||
Prediction Results → Data Formatting → OPC Server Write →
|
||||
Success/Failure Metrics → Notification System
|
||||
```
|
||||
|
||||
### Security Architecture
|
||||
|
||||
#### **Authentication & Authorization**
|
||||
- **Certificate-based OPC Authentication**: Secure industrial communication
|
||||
- **MLFlow API Authentication**: Username/password with secure transmission
|
||||
- **Database Connection Security**: Encrypted connections with credential management
|
||||
- **Kubernetes Secrets Integration**: Secure credential storage and access
|
||||
|
||||
#### **Network Security**
|
||||
- **TLS/SSL Encryption**: Secure communication channels
|
||||
- **Network Isolation**: Kubernetes network policies and service mesh
|
||||
- **Firewall Rules**: Controlled access to external services
|
||||
- **VPN Integration**: Secure remote access and management
|
||||
|
||||
#### **Data Security**
|
||||
- **Data Encryption**: At-rest and in-transit encryption
|
||||
- **Access Control**: Role-based access control (RBAC)
|
||||
- **Audit Logging**: Comprehensive access and operation logging
|
||||
- **Data Retention**: Configurable data lifecycle management
|
||||
|
||||
## 🔄 Workflows
|
||||
|
||||
### 1. Predictions Batch Workflow (`predictions_batch.py`)
|
||||
|
||||
The **PredictionsBatch** workflow is the main entry point for batch prediction pipelines. It orchestrates the complete prediction process and implements a robust data loading and processing pattern.
|
||||
|
||||
#### Purpose
|
||||
- **Batch Prediction Orchestration**: Coordinates data loading and prediction processing
|
||||
- **Data Preparation**: Loads data using custom SQL queries with configurable schemas
|
||||
- **Workflow Delegation**: Delegates actual prediction processing to the PredictionProcess workflow
|
||||
- **Configuration Management**: Handles model configuration, filters, and retention policies
|
||||
|
||||
#### Execution Flow
|
||||
1. **Data Loading**: Executes custom SQL query to load data from PostgreSQL
|
||||
2. **Input Preparation**: Prepares prediction input with metadata and configuration
|
||||
3. **Workflow Delegation**: Spawns PredictionProcess child workflow for actual processing
|
||||
4. **Error Handling**: Implements comprehensive error handling with retry policies
|
||||
|
||||
#### Key Features
|
||||
- **Custom Query Support**: Flexible SQL-based data loading
|
||||
- **Schema Configuration**: Configurable data schema definitions
|
||||
- **Automatic Retry**: Implements Temporal retry policies for fault tolerance
|
||||
- **Timeout Management**: 60-second timeout for all activities
|
||||
- **Comprehensive Error Handling**: Detailed error reporting and notification integration
|
||||
|
||||
#### Input Parameters
|
||||
```json
|
||||
{
|
||||
"schedule_name": "hourly_predictions",
|
||||
"model_name": "temperature_prediction_model",
|
||||
"model_id": "temp_pred_001",
|
||||
"query": "SELECT * FROM sensor_data WHERE timestamp > NOW() - INTERVAL '1 hour'",
|
||||
"schema": {
|
||||
"timestamp": "datetime",
|
||||
"temperature": "float",
|
||||
"humidity": "float"
|
||||
},
|
||||
"table_name": "predictions",
|
||||
"input_filters": {
|
||||
"EMPTY_DATA": {"POLICY": "STOP"}
|
||||
},
|
||||
"mlflow_transform_filters": {
|
||||
"API_ERROR": {"POLICY": "STOP"}
|
||||
},
|
||||
"mlflow_predict_filters": {
|
||||
"API_ERROR": {"POLICY": "STOP"}
|
||||
},
|
||||
"model_retention": 60,
|
||||
"path_priority": ["STOP", "CONTINUE", "REPEAT"],
|
||||
"opc_output_config": {
|
||||
"server_id": "opc_server_1",
|
||||
"tags": ["prediction_output"]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Architecture Diagram
|
||||
```mermaid
|
||||
flowchart LR
|
||||
A[1. load_custom_query] --> B[2. prediction_process 🔃]
|
||||
|
||||
A -.-> Database[(Database)]
|
||||
```
|
||||
|
||||
### 2. Prediction Process Workflow (`prediction_process.py`)
|
||||
|
||||
The **PredictionProcess** workflow implements the core prediction pipeline for ML model inference. It handles data quality validation, MLFlow model interactions, and prediction processing.
|
||||
|
||||
#### Purpose
|
||||
- **Data Quality Validation**: Applies configurable filters for data integrity
|
||||
- **MLFlow Integration**: Manages model transformation and prediction requests
|
||||
- **Response Validation**: Filters MLFlow API responses for quality assurance
|
||||
- **Prediction Export**: Delegates prediction formatting and export operations
|
||||
|
||||
#### Execution Flow
|
||||
1. **Timestamp Retrieval**: Gets the last processed timestamp for incremental processing
|
||||
2. **Input Data Gate**: Applies configured filters for data quality validation
|
||||
3. **Path Decision**: Determines processing path based on filter results
|
||||
4. **MLFlow Transform**: Requests data transformation using MLFlow models
|
||||
5. **Response Validation**: Filters transform responses for quality assurance
|
||||
6. **MLFlow Prediction**: Executes prediction using transformed data
|
||||
7. **Content Validation**: Filters prediction responses for final quality check
|
||||
8. **Export Delegation**: Delegates to FormatAndExportPrediction workflow
|
||||
|
||||
#### Key Features
|
||||
- **Configurable Quality Gates**: Multiple filter types with policy-based configuration
|
||||
- **Flexible Path Handling**: Configurable decision paths (STOP, CONTINUE, REPEAT)
|
||||
- **MLFlow Integration**: Comprehensive model management and inference
|
||||
- **Incremental Processing**: Timestamp-based data processing optimization
|
||||
- **Comprehensive Monitoring**: Detailed metrics and error reporting
|
||||
|
||||
#### Input Parameters
|
||||
```json
|
||||
{
|
||||
"metadata": {
|
||||
"schedule_name": "hourly_predictions",
|
||||
"model_name": "temperature_prediction_model",
|
||||
"model_id": "temp_pred_001",
|
||||
"workflow_name": "predictions_batch"
|
||||
},
|
||||
"data": {...},
|
||||
"schema": {...},
|
||||
"table_name": "predictions",
|
||||
"model_id": "temp_pred_001",
|
||||
"model_name": "temperature_prediction_model",
|
||||
"input_filters": {
|
||||
"EMPTY_DATA": {"POLICY": "STOP"},
|
||||
"SPECIFIC_VARIABLES_NULL_VALUES": {
|
||||
"POLICY": "STOP",
|
||||
"config": {"variables": ["temperature", "humidity"]}
|
||||
}
|
||||
},
|
||||
"mlflow_transform_filters": {
|
||||
"API_ERROR": {"POLICY": "STOP"}
|
||||
},
|
||||
"mlflow_predict_filters": {
|
||||
"API_ERROR": {"POLICY": "STOP"},
|
||||
"NAN_VALUES": {"POLICY": "STOP"}
|
||||
},
|
||||
"model_retention": 60,
|
||||
"path_priority": ["STOP", "CONTINUE", "REPEAT"],
|
||||
"opc_output_config": {...}
|
||||
}
|
||||
```
|
||||
|
||||
#### Architecture Diagram
|
||||
```mermaid
|
||||
flowchart LR
|
||||
A[1. get_last_timestamp] --> B[2. input_gate] --> C[3. request_transform] --> D[4. mlflow_response_gate] --> E[5. mlflow_content_gate] --> F[6. request_predict] --> G[7. mlflow_response_gate] --> H[8. format_and_export_prediction🔃]
|
||||
|
||||
A -.-> Redis[(Redis)]
|
||||
C -.-> MLFlow[MLFlow]
|
||||
F -.-> MLFlow[MLFlow]
|
||||
G -.-> Filters[MLFlow Filters]
|
||||
```
|
||||
|
||||
### 3. Format and Export Prediction Workflow (`format_and_export_prediction.py`)
|
||||
|
||||
The **FormatAndExportPrediction** workflow handles prediction data formatting and export operations to multiple destinations.
|
||||
|
||||
#### Purpose
|
||||
- **Data Formatting**: Formats prediction data for different output destinations
|
||||
- **PostgreSQL Export**: Persists predictions to database with metrics
|
||||
- **OPC Integration**: Writes predictions to OPC servers for real-time access
|
||||
- **Metrics Recording**: Tracks export operations and performance metrics
|
||||
|
||||
#### Execution Flow
|
||||
1. **Path Decision**: Determines formatting path based on configuration
|
||||
2. **Data Formatting**: Formats prediction data for specific output requirements
|
||||
3. **PostgreSQL Export**: Writes formatted predictions to database
|
||||
4. **OPC Export**: Writes predictions to OPC servers
|
||||
5. **Metrics Recording**: Records export performance and success metrics
|
||||
|
||||
#### Key Features
|
||||
- **Flexible Formatting**: Configurable output formats for different destinations
|
||||
- **Multi-Destination Export**: PostgreSQL and OPC server integration
|
||||
- **Performance Monitoring**: Comprehensive metrics for export operations
|
||||
- **Error Handling**: Robust error handling with notification integration
|
||||
|
||||
#### Architecture Diagram
|
||||
```mermaid
|
||||
flowchart LR
|
||||
A[1. format_prediction/format_default_prediction] --> B[2. write_opc_data] --> C[3. export_data_to_postgres] --> D[4. write_metrics]
|
||||
|
||||
A -.-> Format[Data Formatting]
|
||||
B -.-> OPC[OPC Servers]
|
||||
C -.-> PostgreSQL[(PostgreSQL)]
|
||||
D -.-> Prometheus[Prometheus]
|
||||
```
|
||||
|
||||
### 4. Minimal Retrain Workflow (`minimal_retrain.py`)
|
||||
|
||||
The **MinimalRetrain** workflow handles automated model retraining and production model updates.
|
||||
|
||||
#### Purpose
|
||||
- **Model Retraining**: Automates ML model retraining processes
|
||||
- **Production Updates**: Manages production model version updates
|
||||
- **Data Export**: Exports training data for model development
|
||||
- **Quality Assurance**: Ensures model quality before production deployment
|
||||
|
||||
#### Execution Flow
|
||||
1. **Data Loading**: Loads training data using custom queries
|
||||
2. **Model Retraining**: Executes model retraining process
|
||||
3. **Quality Validation**: Validates retrained model performance
|
||||
4. **Production Update**: Updates production model if quality criteria met
|
||||
5. **Data Export**: Exports training data for analysis
|
||||
|
||||
#### Architecture Diagram
|
||||
```mermaid
|
||||
flowchart LR
|
||||
A[1. load_custom_query] --> B[2. retrain_model] --> C[3. update_production_model] --> D[4. export_data_to_postgres]
|
||||
|
||||
A -.-> Database[(Database)]
|
||||
B -.-> MLFlow[MLFlow]
|
||||
C -.-> MLFlow[MLFlow]
|
||||
D -.-> PostgreSQL[(PostgreSQL)]
|
||||
```
|
||||
|
||||
## 📋 Prerequisites
|
||||
|
||||
- Python 3.11+
|
||||
- Temporal server/cluster
|
||||
- PostgreSQL database
|
||||
- MLFlow server
|
||||
- OPC server(s)
|
||||
- MongoDB server (for notifications)
|
||||
|
||||
**Note**: External dependencies must be available either through:
|
||||
- Kubernetes cluster deployment
|
||||
- Docker Compose setup
|
||||
- Cloud-managed services
|
||||
- Local installations
|
||||
|
||||
## 🚀 Installation
|
||||
|
||||
### Local Development Setup
|
||||
|
||||
1. **Clone the repository**
|
||||
```bash
|
||||
git clone <repository-url>
|
||||
cd sientia-dataops-laborious
|
||||
```
|
||||
|
||||
2. **Create virtual environment**
|
||||
```bash
|
||||
python3.11 -m venv venv
|
||||
source ./venv/bin/activate
|
||||
```
|
||||
|
||||
3. **Install dependencies**
|
||||
|
||||
1. **Install github cli**
|
||||
```bash
|
||||
sudo apt update
|
||||
sudo apt install gh -y
|
||||
```
|
||||
|
||||
2. **Authenticate with github**
|
||||
```bash
|
||||
gh auth login
|
||||
```
|
||||
|
||||
3. **Run the install_dependencies.sh script**
|
||||
```bash
|
||||
chmod +x install_dependencies.sh
|
||||
./install_dependencies.sh
|
||||
```
|
||||
|
||||
4. **Create environment configuration file**
|
||||
```bash
|
||||
cp .env.example .env
|
||||
# Edit .env with your connection details
|
||||
```
|
||||
|
||||
5. **Configure external dependencies**
|
||||
|
||||
You'll need to set up port forwarding or connections to external services. For example:
|
||||
|
||||
```bash
|
||||
# Port forwarding from Kubernetes cluster
|
||||
kubectl port-forward svc/postgresql 5432:5432
|
||||
kubectl port-forward svc/mlflow 5000:5000
|
||||
kubectl port-forward svc/mongodb 27017:27017
|
||||
|
||||
# Or connect to external services
|
||||
# Ensure services are accessible on localhost with appropriate ports
|
||||
```
|
||||
|
||||
## 📦 How to Run
|
||||
|
||||
### Running the Laborious Application
|
||||
|
||||
Use the provided script to run the application locally:
|
||||
|
||||
```bash
|
||||
# Make script executable (first time only)
|
||||
chmod +x run_local.sh
|
||||
|
||||
# Run the application
|
||||
./run_local.sh
|
||||
```
|
||||
|
||||
The script will:
|
||||
- Activate the virtual environment
|
||||
- Load environment variables from `.env`
|
||||
- Start the laborious worker application
|
||||
|
||||
### Running Tests and Coverage
|
||||
|
||||
Use the provided script to run tests with coverage:
|
||||
|
||||
```bash
|
||||
# Make script executable (first time only)
|
||||
chmod +x run_coverage.sh
|
||||
|
||||
# Run tests with coverage
|
||||
./run_coverage.sh
|
||||
```
|
||||
|
||||
The script will:
|
||||
- Activate the virtual environment
|
||||
- Run pytest with coverage reporting
|
||||
- Generate HTML coverage report
|
||||
- Open the coverage report in your browser
|
||||
|
||||
### Manual Test Execution
|
||||
|
||||
You can also run tests manually:
|
||||
|
||||
```bash
|
||||
# Activate virtual environment
|
||||
source ./venv/bin/activate
|
||||
|
||||
# Run all tests
|
||||
pytest
|
||||
|
||||
# Run with coverage
|
||||
pytest --cov=laborious --cov-report=html
|
||||
|
||||
# Run specific test categories
|
||||
pytest tests/activities/
|
||||
pytest tests/workflow/
|
||||
```
|
||||
|
||||
### Manual Application Execution
|
||||
|
||||
For manual execution without scripts:
|
||||
|
||||
```bash
|
||||
# Activate virtual environment
|
||||
source ./venv/bin/activate
|
||||
|
||||
# Load environment variables (if using .env file)
|
||||
if [ -f .env ]; then
|
||||
export $(cat .env | grep -v '^#' | xargs)
|
||||
fi
|
||||
|
||||
# Start the laborious worker
|
||||
python -m laborious.worker.worker
|
||||
```
|
||||
|
||||
## 🧪 Testing
|
||||
|
||||
### Test Structure
|
||||
```
|
||||
tests/
|
||||
├── activities/ # Activity implementation tests
|
||||
├── workflow/ # Workflow orchestration tests
|
||||
├── utils/ # Utility function tests
|
||||
└── integration/ # End-to-end workflow tests
|
||||
```
|
||||
|
||||
### Test Execution
|
||||
```bash
|
||||
# Install test dependencies
|
||||
pip install pytest pytest-cov pytest-asyncio
|
||||
|
||||
# Run tests with coverage
|
||||
pytest --cov=laborious --cov-report=html
|
||||
|
||||
# Run specific test modules
|
||||
pytest tests/activities/test_gates.py
|
||||
pytest tests/workflow/test_predictions_batch.py
|
||||
```
|
||||
|
||||
## 📊 Monitoring and Metrics
|
||||
|
||||
The Laborious system exposes comprehensive Prometheus metrics:
|
||||
|
||||
### Application 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
|
||||
|
||||
### MLFlow Metrics
|
||||
- Model transformation and prediction success rates
|
||||
- API response times and error rates
|
||||
- Model retention and versioning metrics
|
||||
|
||||
### Export Metrics
|
||||
- PostgreSQL export operation counts and response times
|
||||
- OPC server write operations and performance
|
||||
- Data quality filter pass/fail rates
|
||||
|
||||
## ⚙️ Configuration
|
||||
|
||||
### Environment Variables
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|----------|-------------|---------|----------|
|
||||
| `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes |
|
||||
| `TEMPORAL_NAMESPACE` | Temporal namespace | `laborious` | No |
|
||||
| `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes |
|
||||
| `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes |
|
||||
| `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 |
|
||||
| `OPC_CONFIG` | OPC server configuration (JSON) | `{}` | No |
|
||||
| `MONGODB_URL` | MongoDB connection URI | `localhost:27017` | Yes |
|
||||
| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
|
||||
| `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No |
|
||||
|
||||
### OPC Configuration
|
||||
|
||||
For multiple OPC servers, use the `OPC_CONFIG` environment variable:
|
||||
|
||||
```json
|
||||
{
|
||||
"opc_server_1": {
|
||||
"url": "opc.tcp://server1:4840",
|
||||
"name": "Server1",
|
||||
"server_uri": "urn:server1:opcua",
|
||||
"cert_path": "/path/to/cert.pem",
|
||||
"private_key_path": "/path/to/key.pem",
|
||||
"server_cert_path": "/path/to/server_cert.pem",
|
||||
"reconnection_interval": 5000
|
||||
},
|
||||
"opc_server_2": {
|
||||
"url": "opc.tcp://server2:4840",
|
||||
"name": "Server2",
|
||||
"server_uri": "urn:server2:opcua",
|
||||
"cert_path": "/path/to/cert.pem",
|
||||
"private_key_path": "/path/to/key.pem",
|
||||
"server_cert_path": "/path/to/server_cert.pem",
|
||||
"reconnection_interval": 5000
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
For single OPC server, use individual environment variables:
|
||||
- `OPC_URL`
|
||||
- `OPC_NAME`
|
||||
- `OPC_SERVER_URI`
|
||||
@@ -83,20 +582,199 @@ For single opc server use:
|
||||
- `OPC_SERVER_CERT_PATH`
|
||||
- `OPC_RECONNECTION_INTERVAL`
|
||||
|
||||
- `TEMPORAL_HOST`
|
||||
- `TEMPORAL_NAMESPACE`
|
||||
### Workflow Configuration
|
||||
|
||||
## Application deployment
|
||||
MongoDB pipeline configuration:
|
||||
|
||||
The application can be deployed using the following command:
|
||||
#### Predictions Batch Workflow
|
||||
```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
|
||||
"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
|
||||
"addr": "ns=2;i=5",
|
||||
"data_type": "double"
|
||||
},
|
||||
{
|
||||
"server_id": "1",
|
||||
"type": "confidence",
|
||||
"addr": "ns=2;i=5",
|
||||
"data_type": "double"
|
||||
}
|
||||
],
|
||||
"input_filters": [
|
||||
{
|
||||
"filter_name": "EMPTY_DATA", # Required filter
|
||||
"policy": "STOP"
|
||||
},
|
||||
{
|
||||
"filter_name": "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"
|
||||
],
|
||||
"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"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 🔧 Development
|
||||
|
||||
### Project Structure
|
||||
```
|
||||
laborious/
|
||||
├── activities/ # Temporal activity implementations
|
||||
│ ├── activities.py # Main activities orchestrator
|
||||
│ ├── gates.py # Data quality gates and filtering
|
||||
│ ├── mlflow.py # MLFlow model operations
|
||||
│ └── opc.py # OPC server operations
|
||||
├── workflow/ # Temporal workflow definitions
|
||||
│ ├── predictions_batch.py # Main batch prediction workflow
|
||||
│ ├── minimal_retrain.py # Model retraining workflow
|
||||
│ └── sub_workflows/ # Sub-workflow implementations
|
||||
│ ├── prediction_process.py # Core prediction workflow
|
||||
│ └── format_and_export_prediction.py # Export workflow
|
||||
├── worker/ # Worker implementation
|
||||
│ └── worker.py # Main worker orchestrator
|
||||
├── utils/ # Utility functions
|
||||
│ ├── connectors_config.py # Database configuration
|
||||
│ ├── filters/ # Data quality filters
|
||||
│ │ ├── conditional_filters.py # Conditional data filters
|
||||
│ │ └── mlflow_filters.py # MLFlow response filters
|
||||
│ └── repository/ # Data access layer
|
||||
│ ├── model_repository.py # MLFlow model operations
|
||||
│ └── opc_repository.py # OPC server operations
|
||||
├── metrics.py # Prometheus metrics definitions
|
||||
└── __init__.py
|
||||
```
|
||||
|
||||
### Adding New Features
|
||||
|
||||
1. **Follow Temporal patterns** for new workflows and activities
|
||||
2. **Add comprehensive docstrings** for all public methods
|
||||
3. **Include Prometheus metrics** for monitoring
|
||||
4. **Add unit tests** for new functionality
|
||||
5. **Update this README** with new features and configuration
|
||||
|
||||
## 🐛 Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
1. **Temporal Connection Failures**
|
||||
- Verify Temporal server is running and accessible
|
||||
- Check namespace configuration and permissions
|
||||
- Review server logs for connection issues
|
||||
|
||||
2. **MLFlow Connection Issues**
|
||||
- Verify MLFlow server is running and accessible
|
||||
- Check authentication credentials and permissions
|
||||
- Ensure model names and versions exist
|
||||
|
||||
3. **Database Connection Issues**
|
||||
- Verify PostgreSQL service is running
|
||||
- Check connection credentials and network access
|
||||
- Ensure proper connection pool configuration
|
||||
|
||||
4. **OPC Connection Failures**
|
||||
- Verify OPC server is accessible
|
||||
- Check certificate and key file paths
|
||||
- Review OPC server logs for connection issues
|
||||
|
||||
5. **Workflow Execution Failures**
|
||||
- Review activity error logs and notifications
|
||||
- Check data quality filter configurations
|
||||
- Verify input data format and required fields
|
||||
|
||||
### Debug Mode
|
||||
|
||||
Enable debug logging by setting the log level:
|
||||
```bash
|
||||
helm upgrade --install sientia-dataops-laborious sientia/sientia-module -n sientia --create-namespace -f ./values.yaml
|
||||
export LOG_LEVEL=DEBUG
|
||||
```
|
||||
|
||||
#PR shortcut
|
||||
```
|
||||
git log origin/main..HEAD --no-merges > git_log
|
||||
```
|
||||
Prompt:
|
||||
Write a summary of PR changes in markdown. Be objective and direct. Write to file
|
||||
## ⚡ Performance Tuning
|
||||
|
||||
### Key Parameters
|
||||
|
||||
- **Worker Concurrency**: Adjust `max_concurrent_workflow_tasks` and `max_concurrent_activities`
|
||||
- **Connection Pools**: Optimize database connection pool sizes
|
||||
- **Model Retention**: Configure MLFlow model retention based on requirements
|
||||
- **Batch Sizes**: Adjust data processing batch sizes for optimal throughput
|
||||
|
||||
### Scaling Considerations
|
||||
|
||||
- **Horizontal Scaling**: Deploy multiple worker instances
|
||||
- **Task Queue Distribution**: Use multiple task queues for different workflow types
|
||||
- **Database Performance**: Optimize indexes and connection pooling
|
||||
- **MLFlow Performance**: Configure appropriate model serving resources
|
||||
|
||||
## 🤝 Contributing
|
||||
|
||||
1. Fork the repository
|
||||
2. Create a feature branch
|
||||
3. Make your changes with comprehensive testing
|
||||
4. Update documentation and docstrings
|
||||
5. Submit a pull request
|
||||
|
||||
### Code Quality Standards
|
||||
|
||||
- Follow PEP 8 style guidelines
|
||||
- Include comprehensive docstrings for all public methods
|
||||
- Maintain test coverage above 80%
|
||||
- Use type hints where appropriate
|
||||
- Follow Temporal.io best practices
|
||||
|
||||
## 📄 License
|
||||
|
||||
This project is licensed under the terms specified in the LICENSE file.
|
||||
|
||||
## 🆘 Support
|
||||
|
||||
For support and questions:
|
||||
- Check the troubleshooting section above
|
||||
- Review the metrics and logs for error patterns
|
||||
- Open an issue in the project repository
|
||||
- Contact the development team
|
||||
|
||||
---
|
||||
|
||||
**Note**: The Laborious system is designed for production use in industrial ML environments. Ensure proper security configuration and network isolation for production deployments.
|
||||
@@ -1 +0,0 @@
|
||||
pytest --cov=laborious --cov-report=html && xdg-open htmlcov/index.html
|
||||
@@ -1,82 +0,0 @@
|
||||
version: '3.8'
|
||||
|
||||
services:
|
||||
postgres:
|
||||
image: postgres:15
|
||||
container_name: postgres
|
||||
environment:
|
||||
POSTGRES_USER: sientia
|
||||
POSTGRES_PASSWORD: sientia
|
||||
POSTGRES_DB: sientia
|
||||
ports:
|
||||
- "5432:5432"
|
||||
volumes:
|
||||
- ./postgres_data:/var/lib/postgresql/data
|
||||
networks:
|
||||
- sientia-network
|
||||
|
||||
zookeeper:
|
||||
image: confluentinc/cp-zookeeper:7.5.1
|
||||
container_name: zookeeper
|
||||
environment:
|
||||
ZOOKEEPER_CLIENT_PORT: 2181
|
||||
ZOOKEEPER_TICK_TIME: 2000
|
||||
ports:
|
||||
- "2181:2181"
|
||||
networks:
|
||||
- sientia-network
|
||||
|
||||
kafka:
|
||||
image: confluentinc/cp-kafka:7.5.1
|
||||
container_name: kafka
|
||||
depends_on:
|
||||
- zookeeper
|
||||
ports:
|
||||
- "9092:9092"
|
||||
- "29092:29092"
|
||||
environment:
|
||||
KAFKA_BROKER_ID: 1
|
||||
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
|
||||
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
|
||||
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
|
||||
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
|
||||
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
|
||||
networks:
|
||||
- sientia-network
|
||||
|
||||
kafka-ui:
|
||||
image: provectuslabs/kafka-ui:latest
|
||||
container_name: kafka-ui
|
||||
ports:
|
||||
- "8080:8080"
|
||||
environment:
|
||||
KAFKA_CLUSTERS_0_NAME: local
|
||||
KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
|
||||
networks:
|
||||
- sientia-network
|
||||
|
||||
simulator:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: simulator/Dockerfile
|
||||
args:
|
||||
GIT_REPO: ${SIMULATOR_GIT_REPO}
|
||||
GIT_BRANCH: ${SIMULATOR_GIT_BRANCH}
|
||||
container_name: simulator
|
||||
ports:
|
||||
- "4840:4840"
|
||||
depends_on:
|
||||
- kafka
|
||||
networks:
|
||||
- sientia-network
|
||||
env_file:
|
||||
- .env
|
||||
|
||||
|
||||
networks:
|
||||
sientia-network:
|
||||
driver: bridge
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
driver: local
|
||||
@@ -1,34 +0,0 @@
|
||||
{
|
||||
"schedule_name": "scouter-opcua-pipeline",
|
||||
"model_name": "Demo Model",
|
||||
"model_id": 1,
|
||||
"query": "SELECT * FROM sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
|
||||
"schema": "sientia_data",
|
||||
"table_name": "predictions",
|
||||
"retention_time": 3600,
|
||||
"model_retention": 120,
|
||||
"path_priority": ["STOP", "CONTINUE", "REPEAT"],
|
||||
"input_filters": {
|
||||
"SPECIFIC_VARIABLES_NULL_VALUES": {
|
||||
"POLICY": "STOP",
|
||||
"VARIABLES": ["Counter"]
|
||||
},
|
||||
"EMPTY_DATA": {
|
||||
"POLICY": "STOP"
|
||||
}
|
||||
},
|
||||
"mlflow_transform_filters": {
|
||||
"API_ERROR": {
|
||||
"POLICY": "CONTINUE"
|
||||
},
|
||||
"NAN_VALUES": {
|
||||
"POLICY": "CONTINUE"
|
||||
}
|
||||
},
|
||||
"mlflow_predict_filters": {
|
||||
"API_ERROR": {
|
||||
"POLICY": "CONTINUE"
|
||||
}
|
||||
},
|
||||
"opc_output_config": {}
|
||||
}
|
||||
@@ -11,13 +11,52 @@ with workflow.unsafe.imports_passed_through():
|
||||
|
||||
|
||||
class Activities(Postgres, MLFlow, Gates, OPC):
|
||||
"""
|
||||
Main activities orchestrator for the Laborious system.
|
||||
|
||||
This class combines functionality from multiple activity classes to provide
|
||||
a unified interface for all workflow operations. It manages database connections,
|
||||
MLFlow model interactions, data quality validation, and OPC server communications.
|
||||
|
||||
The class implements multiple inheritance to combine specialized functionality:
|
||||
- Postgres: Database operations and data persistence
|
||||
- MLFlow: Model inference and transformation operations
|
||||
- Gates: Data quality validation and filtering mechanisms
|
||||
- OPC: Real-time data export to OPC servers
|
||||
|
||||
Attributes:
|
||||
postgres_config (dict): PostgreSQL connection configuration
|
||||
mlflow_config (dict): MLFlow server configuration
|
||||
opc_config (dict): OPC server configuration
|
||||
logger (Logger): Logging and observability instance
|
||||
notification_handler (NotificationHandler): Notification management instance
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
postgres_config: dict[str, Any],
|
||||
mlflow_config: dict[str, Any],
|
||||
opc_config: dict[str, Any],
|
||||
logger: Logger, notification_handler: NotificationHandler):
|
||||
logger: Logger,
|
||||
notification_handler: NotificationHandler):
|
||||
"""
|
||||
Initialize the Activities orchestrator with all required configurations.
|
||||
|
||||
This constructor initializes all parent classes with their respective
|
||||
configurations and sets up the foundation for all activity operations.
|
||||
|
||||
Args:
|
||||
postgres_config: PostgreSQL connection configuration dictionary
|
||||
Required keys: host, port, user, password, dbname, min_connections, max_connections
|
||||
mlflow_config: MLFlow server configuration dictionary
|
||||
Required keys: host, port, username, password
|
||||
opc_config: OPC server configuration dictionary
|
||||
Can contain multiple server configurations
|
||||
logger: Logger instance for observability and debugging
|
||||
notification_handler: Notification handler for alerts and monitoring
|
||||
|
||||
Raises:
|
||||
Exception: If any parent class initialization fails
|
||||
"""
|
||||
# Initialize parent classes
|
||||
Postgres.__init__(self, host=postgres_config['host'],
|
||||
port=postgres_config['port'],
|
||||
@@ -45,5 +84,16 @@ class Activities(Postgres, MLFlow, Gates, OPC):
|
||||
notification_handler=notification_handler)
|
||||
|
||||
async def shutdown(self):
|
||||
"""
|
||||
Gracefully shutdown all activities and clean up resources.
|
||||
|
||||
This method ensures proper cleanup of all resources including:
|
||||
- PostgreSQL connection pools
|
||||
- OPC server connections
|
||||
- Any other resources that need explicit cleanup
|
||||
|
||||
The method should be called before the application terminates to ensure
|
||||
proper resource cleanup and prevent resource leaks.
|
||||
"""
|
||||
Postgres.close(self)
|
||||
await OPC.shutdown(self)
|
||||
|
||||
@@ -17,6 +17,7 @@ with workflow.unsafe.imports_passed_through():
|
||||
from pandas import DataFrame
|
||||
from laborious import metrics
|
||||
|
||||
# Input filter function mappings
|
||||
input_filter_functions = {
|
||||
'SPECIFIC_VARIABLES_NULL_VALUES': filter_specific_variables_null_values,
|
||||
'EMPTY_DATA': filter_empty_data,
|
||||
@@ -27,6 +28,7 @@ input_filter_functions = {
|
||||
}
|
||||
}
|
||||
|
||||
# MLFlow response filter function mappings
|
||||
mlflow_response_filter_functions = {
|
||||
'API_ERROR': api_error_filter,
|
||||
'path_confidence': {
|
||||
@@ -36,6 +38,7 @@ mlflow_response_filter_functions = {
|
||||
},
|
||||
}
|
||||
|
||||
# MLFlow content filter function mappings
|
||||
mlflow_content_filter_functions = {
|
||||
'NAN_VALUES': nan_values_filter,
|
||||
'path_confidence': {
|
||||
@@ -47,24 +50,72 @@ mlflow_content_filter_functions = {
|
||||
|
||||
|
||||
class Gates(BaseActivity):
|
||||
"""
|
||||
Data quality gates and filtering activities for the Laborious system.
|
||||
|
||||
This class implements comprehensive data quality validation and filtering
|
||||
mechanisms that can be applied at different stages of the prediction pipeline.
|
||||
It provides configurable filters with policy-based decision making to ensure
|
||||
data integrity and quality throughout the ML workflow.
|
||||
|
||||
The class supports multiple filter types and implements a flexible policy
|
||||
system that can be configured for different validation requirements. Each
|
||||
filter returns a path decision (STOP, CONTINUE, REPEAT) along with confidence
|
||||
scores and detailed comments for monitoring and debugging.
|
||||
|
||||
Attributes:
|
||||
input_filter_functions (dict): Mapping of input filter names to functions
|
||||
mlflow_response_filter_functions (dict): Mapping of MLFlow response filter names to functions
|
||||
mlflow_content_filter_functions (dict): Mapping of MLFlow content filter names to functions
|
||||
"""
|
||||
|
||||
def __init__(self, logger: Logger, notification_handler: NotificationHandler):
|
||||
"""
|
||||
Initialize data quality gates with logging and notification capabilities.
|
||||
|
||||
Args:
|
||||
logger: Logger instance for observability and debugging
|
||||
notification_handler: Notification handler for alerts and monitoring
|
||||
|
||||
Raises:
|
||||
Exception: If BaseActivity initialization fails
|
||||
"""
|
||||
BaseActivity.__init__(
|
||||
self, logger, notification_handler, set_error_counter=True)
|
||||
|
||||
@activity.defn(name="input_gate")
|
||||
async def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
|
||||
"""
|
||||
Filters the data based on the filters. The return value is a tuple with the first element
|
||||
being the policy and the second element being the confidence status.
|
||||
Apply input data quality filters and validation.
|
||||
|
||||
This activity validates input data quality using configurable filters
|
||||
before proceeding with ML operations. It applies multiple filter types
|
||||
and returns a path decision based on the filter results and configured
|
||||
policies.
|
||||
|
||||
The method implements a comprehensive filtering system that:
|
||||
1. Applies configured filters to input data
|
||||
2. Evaluates filter results against policy configurations
|
||||
3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT)
|
||||
4. Provides confidence scores and detailed comments
|
||||
5. Handles errors gracefully with notification integration
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- filters (dict): The filters to apply.
|
||||
The key is the filter name and the value is the filter configuration.
|
||||
- data (dict[str, Any]): The data to filter.
|
||||
- path_priority (list[str]): The path priority.
|
||||
input_data: Configuration and data for input validation
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- filters (dict): Filter configuration and policies
|
||||
- data (dict): Input data to validate
|
||||
- path_priority (list[str]): Priority order for path decisions
|
||||
|
||||
Returns:
|
||||
tuple[str | None, int, str]: (policy, confidence, comments) based in priority
|
||||
list and filter configuration and functions.
|
||||
tuple: (path_flag, confidence, comment)
|
||||
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
|
||||
- confidence (int): Confidence score for the decision
|
||||
- comment (str): Detailed explanation of the decision
|
||||
|
||||
Raises:
|
||||
Exception: If filter execution fails or configuration is invalid
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
|
||||
@@ -81,6 +132,7 @@ class Gates(BaseActivity):
|
||||
self.debug(f"Input data:\n {data}", metadata)
|
||||
self.debug(f"Filters: {filters}", metadata)
|
||||
|
||||
# Apply each configured filter
|
||||
for fil, config in filters.items():
|
||||
if fil not in input_filter_functions:
|
||||
self.error(f"Filter {fil} not found", metadata)
|
||||
@@ -113,20 +165,37 @@ class Gates(BaseActivity):
|
||||
@activity.defn(name="mlflow_response_gate")
|
||||
async def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
|
||||
"""
|
||||
Filters the data based on the mlflow response filters.
|
||||
The return value is a tuple with the first element
|
||||
being the policy and the second element being the confidence status.
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- filters (dict): The filter configuration to apply.
|
||||
- data (dict[str, Any]): The data to filter.
|
||||
- path_priority (list[str]): The path priority list.
|
||||
- type (str): The type of the gate.
|
||||
Returns:
|
||||
tuple[str | None, int, str]: (policy, confidence, comments) based in priority list
|
||||
and filter configuration and functions.
|
||||
"""
|
||||
Validate MLFlow API response quality and integrity.
|
||||
|
||||
This activity validates MLFlow API responses to ensure they meet quality
|
||||
standards before proceeding with further processing. It applies response-specific
|
||||
filters and determines appropriate path decisions based on response quality.
|
||||
|
||||
The method implements response validation that:
|
||||
1. Applies MLFlow response-specific filters
|
||||
2. Evaluates API response quality and integrity
|
||||
3. Determines path decisions based on response validation results
|
||||
4. Provides confidence scores and detailed validation comments
|
||||
5. Handles API errors and response validation failures
|
||||
|
||||
Args:
|
||||
input_data: Configuration and data for response validation
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- filters (dict): Response filter configuration and policies
|
||||
- data (dict): MLFlow API response data to validate
|
||||
- type (str): Type of MLFlow operation (transform, predict)
|
||||
- path_priority (list[str]): Priority order for path decisions
|
||||
|
||||
Returns:
|
||||
tuple: (path_flag, confidence, comment)
|
||||
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
|
||||
- confidence (int): Confidence score for the decision
|
||||
- comment (str): Detailed explanation of the decision
|
||||
|
||||
Raises:
|
||||
Exception: If response validation fails or configuration is invalid
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
self.info("Performing mlflow response gate...", metadata)
|
||||
|
||||
@@ -143,6 +212,7 @@ class Gates(BaseActivity):
|
||||
comments = []
|
||||
for fil, config in filters.items():
|
||||
if fil not in mlflow_response_filter_functions:
|
||||
self.error(f"Filter {fil} not found", metadata)
|
||||
continue
|
||||
try:
|
||||
if mlflow_response_filter_functions[fil](data, config):
|
||||
@@ -180,20 +250,37 @@ class Gates(BaseActivity):
|
||||
@activity.defn(name="mlflow_content_gate")
|
||||
async def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
|
||||
"""
|
||||
Filters the data based on the mlflow content filters.
|
||||
The return value is a tuple with the first element
|
||||
being the policy and the second element being the confidence status.
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- filters (dict): The filter configuration to apply.
|
||||
- data (dict[str, Any]): The data to filter.
|
||||
- path_priority (list[str]): The path priority list.
|
||||
- type (str): The type of the gate.
|
||||
Returns:
|
||||
tuple[str | None, int, str]: (policy, confidence, comments) based in priority
|
||||
list and filter configuration and functions.
|
||||
"""
|
||||
Validate MLFlow prediction content quality and integrity.
|
||||
|
||||
This activity validates the content of MLFlow predictions to ensure they
|
||||
meet quality standards before export and persistence. It applies content-specific
|
||||
filters and determines appropriate path decisions based on content quality.
|
||||
|
||||
The method implements content validation that:
|
||||
1. Applies MLFlow content-specific filters
|
||||
2. Evaluates prediction content quality and integrity
|
||||
3. Determines path decisions based on content validation results
|
||||
4. Provides confidence scores and detailed validation comments
|
||||
5. Handles content validation failures and quality issues
|
||||
|
||||
Args:
|
||||
input_data: Configuration and data for content validation
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- filters (dict): Content filter configuration and policies
|
||||
- data (dict): MLFlow prediction content to validate
|
||||
- type (str): Type of MLFlow operation (transform, predict)
|
||||
- path_priority (list[str]): Priority order for path decisions
|
||||
|
||||
Returns:
|
||||
tuple: (path_flag, confidence, comment)
|
||||
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
|
||||
- confidence (int): Confidence score for the decision
|
||||
- comment (str): Detailed explanation of the decision
|
||||
|
||||
Raises:
|
||||
Exception: If content validation fails or configuration is invalid
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
self.info("Performing mlflow content gate...", metadata)
|
||||
|
||||
@@ -242,48 +329,150 @@ class Gates(BaseActivity):
|
||||
self.info("Nothing was filtered by the mlflow content gate", metadata)
|
||||
return None, 0, ""
|
||||
|
||||
def get_prediction_store_policy(self,
|
||||
prediction_store_policy: str,
|
||||
metadata: dict[str, Any]) -> tuple[str, int]:
|
||||
"""
|
||||
Parse and validate prediction store policy configuration.
|
||||
|
||||
This method parses prediction store policy strings in the format 'type:value'
|
||||
and validates them against allowed policy types and values. It provides
|
||||
sensible defaults for invalid configurations and logs policy validation
|
||||
failures for operational monitoring.
|
||||
|
||||
Supported Policy Types:
|
||||
- 'lts': Latest timestamp - sorts data by timestamp descending
|
||||
- 'erl': Earliest timestamp - sorts data by timestamp ascending
|
||||
|
||||
Args:
|
||||
prediction_store_policy (str): Policy string in format 'type:value'
|
||||
metadata (dict[str, Any]): Context metadata for logging and notifications
|
||||
|
||||
Returns:
|
||||
tuple[str, int]: (policy_type, policy_value)
|
||||
- policy_type (str): Validated policy type ('lts' or 'erl')
|
||||
- policy_value (int): Number of rows to retain
|
||||
"""
|
||||
policy_elements = prediction_store_policy.split(':')
|
||||
|
||||
if len(policy_elements) < 2:
|
||||
self.error(
|
||||
f"Invalid prediction store policy: {prediction_store_policy}, using default policy", metadata)
|
||||
return 'lts', 1
|
||||
|
||||
policy_type = policy_elements[0]
|
||||
policy_value = policy_elements[1]
|
||||
|
||||
# If the policy_type is not lts or erl, we use the default policy
|
||||
# If the policty_value is not a number or 0, we use the default policy
|
||||
if policy_type not in ['lts', 'erl'] or not policy_value.isdigit() or int(policy_value) == 0:
|
||||
self.error(
|
||||
f"Invalid prediction store policy: {prediction_store_policy}, using default policy", metadata)
|
||||
return 'lts', 1
|
||||
|
||||
return policy_type, int(policy_value)
|
||||
|
||||
@activity.defn(name="format_prediction")
|
||||
async def format_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]:
|
||||
"""
|
||||
Formats the prediction data.
|
||||
Format prediction data according to configured storage policies.
|
||||
|
||||
This method formats prediction data for storage and export operations.
|
||||
It applies timestamp-based sorting policies, adds metadata fields,
|
||||
and ensures data consistency before persistence. The method supports
|
||||
multiple storage policies for flexible data retention strategies.
|
||||
|
||||
Storage Policies:
|
||||
- 'lts:N': Latest timestamp - retains N most recent predictions
|
||||
- 'erl:N': Earliest timestamp - retains N oldest predictions
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- data (dict[str, Any]): The data to format.
|
||||
- timestamp (str): The timestamp of the data.
|
||||
- model_id (str): The id of the model.
|
||||
- prediction_confidence (float): The confidence of the prediction.
|
||||
input_data (dict): Input data containing:
|
||||
- data (dict[str, Any]): Raw prediction data to format
|
||||
- timestamp (str): Default timestamp if data lacks timestamp column
|
||||
- model_id (str): Unique identifier for the ML model
|
||||
- prediction_confidence (float): Confidence score for the prediction
|
||||
- prediction_store_policy (str): Storage policy in format 'type:value'
|
||||
|
||||
Returns:
|
||||
dict: The formatted data.
|
||||
dict: Formatted prediction data ready for storage and export
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
prediction_store_policy = input_data['prediction_store_policy']
|
||||
self.info("Formatting prediction...", metadata)
|
||||
|
||||
data = DataFrame(input_data['data'])
|
||||
data['timestamp'] = input_data['timestamp']
|
||||
|
||||
self.debug(
|
||||
f"Prediction store policy: {prediction_store_policy}", metadata)
|
||||
|
||||
policy_type, policy_value = self.get_prediction_store_policy(
|
||||
prediction_store_policy, metadata)
|
||||
|
||||
# If data has no timestamp, we use the default timestamp and not sort the data
|
||||
if 'timestamp' not in data.columns:
|
||||
self.warning(
|
||||
"Data has no timestamp, using default timestamp", metadata)
|
||||
data['timestamp'] = input_data['timestamp']
|
||||
else:
|
||||
self.debug(
|
||||
"Data has timestamp, sorting data by timestamp", metadata)
|
||||
|
||||
# If policy_type is lts, we need to sort the data by timestamp descending and take the first policy_value rows
|
||||
if policy_type == 'lts':
|
||||
self.debug(
|
||||
"Sorting data by timestamp descending", metadata)
|
||||
data = data.sort_values(by='timestamp', ascending=False)
|
||||
# If policy_type is erl, we need to sort the data by timestamp ascending and take the first policy_value rows
|
||||
elif policy_type == 'erl':
|
||||
self.debug(
|
||||
"Sorting data by timestamp ascending", metadata)
|
||||
data = data.sort_values(by='timestamp', ascending=True)
|
||||
else:
|
||||
self.error(
|
||||
f"Invalid policy type: {policy_type}, using default policy", metadata)
|
||||
raise ValueError(
|
||||
f"Invalid policy type: {policy_type}")
|
||||
|
||||
data = data.head(int(policy_value))
|
||||
|
||||
data['model_id'] = input_data['model_id']
|
||||
data['prediction_confidence'] = input_data['prediction_confidence']
|
||||
data['prediction_status'] = 'Good'
|
||||
data['comments'] = ""
|
||||
data = data.sort_values(by='timestamp')
|
||||
data = data.sort_values(by='timestamp', ascending=False)
|
||||
data = data.reset_index(drop=True)
|
||||
|
||||
self.info(f"Prediction formatted: {data.size} rows", metadata)
|
||||
self.debug(f"Prediction data: {data.to_string()}", metadata)
|
||||
|
||||
return data.to_dict()
|
||||
|
||||
@activity.defn(name="format_default_prediction")
|
||||
async def format_default_prediction(self, input_data: dict[str, Any]) -> dict[Any, Any]:
|
||||
"""
|
||||
Creates and formats the default prediction data, with zero value in prediction,
|
||||
and usefull information in the other fields.
|
||||
Create and format default prediction data for error conditions.
|
||||
|
||||
This method generates default prediction data when the main prediction
|
||||
pipeline encounters errors or quality issues. It creates a standardized
|
||||
data structure with zero values for predictions and useful metadata
|
||||
for operational monitoring and debugging.
|
||||
|
||||
The default prediction serves as a fallback mechanism to:
|
||||
1. Maintain data pipeline continuity during failures
|
||||
2. Provide operational visibility into prediction quality issues
|
||||
3. Enable downstream systems to handle error conditions gracefully
|
||||
4. Support debugging and troubleshooting efforts
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- timestamp (str): The timestamp of the data.
|
||||
- model_id (str): The id of the model.
|
||||
- prediction_confidence (float): The confidence of the prediction.
|
||||
- comment (str): The comment of the prediction.
|
||||
input_data (dict): Input data containing:
|
||||
- timestamp (str): Timestamp for the default prediction
|
||||
- model_id (str): Unique identifier for the ML model
|
||||
- prediction_confidence (float): Confidence score (typically low for errors)
|
||||
- comment (str): Error description or operational comment
|
||||
|
||||
Returns:
|
||||
dict: The formatted data.
|
||||
dict: Formatted default prediction data with error indicators
|
||||
"""
|
||||
|
||||
metadata = input_data['metadata']
|
||||
@@ -305,12 +494,25 @@ class Gates(BaseActivity):
|
||||
@activity.defn(name="get_last_timestamp")
|
||||
async def get_last_timestamp(self, input_data: dict[str, Any]) -> str:
|
||||
"""
|
||||
Gets the last timestamp of the data.
|
||||
Extract the most recent timestamp from prediction data.
|
||||
|
||||
This method analyzes prediction data to find the latest timestamp,
|
||||
enabling incremental processing and data continuity tracking.
|
||||
It handles empty datasets gracefully by returning the current time
|
||||
as a fallback timestamp.
|
||||
|
||||
The method is essential for:
|
||||
1. Incremental data processing workflows
|
||||
2. Data continuity validation
|
||||
3. Timestamp-based data loading optimization
|
||||
4. Workflow execution tracking
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- data (dict[str, Any]): The data to get the last timestamp from.
|
||||
input_data (dict): Input data containing:
|
||||
- data (dict[str, Any]): Prediction data to analyze
|
||||
|
||||
Returns:
|
||||
str: The last timestamp of the data.
|
||||
str: Formatted timestamp string in UTC with timezone
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
|
||||
@@ -334,10 +536,25 @@ class Gates(BaseActivity):
|
||||
@activity.defn(name="write_metrics")
|
||||
async def write_metrics(self, input_data: dict[str, Any]):
|
||||
"""
|
||||
Write metrics to the database.
|
||||
input_data:
|
||||
metadata: dict[str, Any]
|
||||
prediction: dict[str, Any]
|
||||
Write prediction performance metrics to Prometheus monitoring system.
|
||||
|
||||
This method records comprehensive metrics for prediction operations,
|
||||
enabling operational monitoring, performance analysis, and alerting.
|
||||
It tracks prediction counts, confidence levels, and response times
|
||||
for each model and pipeline combination.
|
||||
|
||||
Metrics Recorded:
|
||||
1. Prediction Count: Incremental counter for successful predictions
|
||||
2. Confidence Monitor: Current confidence level for predictions
|
||||
3. Response Time Monitor: Histogram of prediction response times
|
||||
|
||||
Args:
|
||||
input_data (dict): Input data containing:
|
||||
- metadata (dict[str, Any]): Workflow execution metadata
|
||||
- prediction (dict[str, Any]): Prediction data with metrics
|
||||
|
||||
Raises:
|
||||
Exception: If metrics writing fails or configuration is invalid
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
prediction = DataFrame(input_data['prediction'])
|
||||
|
||||
@@ -15,8 +15,40 @@ with workflow.unsafe.imports_passed_through():
|
||||
|
||||
|
||||
class MLFlow(BaseActivity):
|
||||
"""
|
||||
MLFlow integration activities for model inference operations.
|
||||
|
||||
This class provides activities for interacting with MLFlow models, including
|
||||
data transformation and prediction operations. It handles authentication,
|
||||
data preprocessing, and model management with configurable retention policies.
|
||||
|
||||
The class implements comprehensive error handling and logging for all
|
||||
MLFlow operations, ensuring reliable model inference in production environments.
|
||||
|
||||
Attributes:
|
||||
mlflow_host (str): MLFlow server hostname
|
||||
mlflow_port (int): MLFlow server port
|
||||
mlflow_username (str): MLFlow authentication username
|
||||
mlflow_password (str): MLFlow authentication password
|
||||
model_monitoring_repository (MLFlowRepository): Repository for MLFlow operations
|
||||
"""
|
||||
|
||||
def __init__(self, mlflow_host: str, mlflow_port: int, mlflow_username: str,
|
||||
mlflow_password: str, logger: Logger, notification_handler: NotificationHandler):
|
||||
"""
|
||||
Initialize MLFlow activities with server configuration.
|
||||
|
||||
Args:
|
||||
mlflow_host: MLFlow server hostname or IP address
|
||||
mlflow_port: MLFlow server port number
|
||||
mlflow_username: Username for MLFlow authentication
|
||||
mlflow_password: Password for MLFlow authentication
|
||||
logger: Logger instance for observability and debugging
|
||||
notification_handler: Notification handler for alerts and monitoring
|
||||
|
||||
Raises:
|
||||
Exception: If MLFlowRepository initialization fails
|
||||
"""
|
||||
BaseActivity.__init__(
|
||||
self, logger, notification_handler, set_error_counter=True)
|
||||
self.mlflow_host = mlflow_host
|
||||
@@ -31,14 +63,32 @@ class MLFlow(BaseActivity):
|
||||
@activity.defn(name="request_transform")
|
||||
async def request_transform(self, input_data: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Access MLFlow model to get the transformed data.
|
||||
Transform input data using MLFlow models.
|
||||
|
||||
This activity processes input data through MLFlow model transformation,
|
||||
including data preprocessing, format conversion, and validation. It handles
|
||||
data deduplication, pivoting, and cleanup to ensure optimal model performance.
|
||||
|
||||
The transformation process includes:
|
||||
1. Data deduplication based on variable and timestamp
|
||||
2. Data pivoting for model input format
|
||||
3. Null value handling and cleanup
|
||||
4. MLFlow model transformation request
|
||||
5. Response validation and logging
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- data (dict[str, Any]): The data to transform.
|
||||
- model_name (str): The name of the model.
|
||||
- model_retention (int): The retention time of the model, in minutes.
|
||||
input_data: Configuration and data for transformation
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- data (dict): Input data for transformation
|
||||
- model_name (str): Name of the MLFlow model to use
|
||||
- model_retention (int): Model retention period in minutes
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: The transformed data.
|
||||
dict: Transformed data from MLFlow model
|
||||
|
||||
Raises:
|
||||
Exception: If transformation fails or MLFlow model is unavailable
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
self.info('Transforming data...', metadata)
|
||||
@@ -54,16 +104,18 @@ class MLFlow(BaseActivity):
|
||||
subset=['variable', 'timestamp'], keep='first'
|
||||
)
|
||||
|
||||
# Pivot data for model input format
|
||||
data = data.pivot(
|
||||
index='timestamp', columns='variable',
|
||||
values='value')
|
||||
data.fillna(np.nan, inplace=True)
|
||||
data.reset_index(inplace=True)
|
||||
# data.reset_index(inplace=True)
|
||||
data.columns.name = None
|
||||
|
||||
self.debug("Processed input data:", metadata)
|
||||
self.debug(data, metadata)
|
||||
|
||||
# Request transformation from MLFlow model
|
||||
response_data = self.model_monitoring_repository.transform(
|
||||
model_name, data, model_retention)
|
||||
|
||||
@@ -77,14 +129,32 @@ class MLFlow(BaseActivity):
|
||||
@activity.defn(name="request_predict")
|
||||
async def request_predict(self, input_data: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Access MLFlow model to get the predicted data.
|
||||
Execute predictions using MLFlow models.
|
||||
|
||||
This activity performs ML model inference using MLFlow models with the
|
||||
transformed data. It handles data format conversion, null value processing,
|
||||
and model prediction requests with comprehensive error handling.
|
||||
|
||||
The prediction process includes:
|
||||
1. Data format validation and cleanup
|
||||
2. Null value handling for model compatibility
|
||||
3. MLFlow model prediction request
|
||||
4. Response validation and logging
|
||||
5. Performance monitoring and metrics
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- data (dict[str, Any]): The data to predict.
|
||||
- model_name (str): The name of the model.
|
||||
- model_retention (int): The retention time of the model, in minutes.
|
||||
input_data: Configuration and data for prediction
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- data (dict): Transformed data for prediction
|
||||
- model_name (str): Name of the MLFlow model to use
|
||||
- model_retention (int): Model retention period in minutes
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: The predicted data.
|
||||
dict: Prediction results from MLFlow model
|
||||
|
||||
Raises:
|
||||
Exception: If prediction fails or MLFlow model is unavailable
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
self.info('Predicting data...', metadata)
|
||||
@@ -94,26 +164,51 @@ class MLFlow(BaseActivity):
|
||||
|
||||
self.debug(data, metadata)
|
||||
|
||||
# Convert numpy.nan to None for model compatibility
|
||||
data.replace(np.nan, None, inplace=True)
|
||||
|
||||
# Request prediction from MLFlow model
|
||||
response_data = self.model_monitoring_repository.predict(
|
||||
model_name, data, model_retention)
|
||||
|
||||
self.debug("Prediction response data:", metadata)
|
||||
self.debug(json.dumps(response_data, indent=4), metadata)
|
||||
|
||||
self.info("Prediction completed successfully", metadata)
|
||||
self.info("Data predicted successfully", metadata)
|
||||
|
||||
return response_data
|
||||
|
||||
@activity.defn(name="retrain_model")
|
||||
async def retrain_model(self, input_data: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Retrain the model.
|
||||
Retrain MLFlow models with updated training data.
|
||||
|
||||
This activity orchestrates the complete model retraining process,
|
||||
including data preparation, model retraining execution, and result
|
||||
validation. It handles data preprocessing, column cleanup, and
|
||||
comprehensive error handling for production model management.
|
||||
|
||||
The retraining process includes:
|
||||
1. Data timestamp extraction and validation
|
||||
2. Column cleanup and data preparation
|
||||
3. Data pivoting for model input format
|
||||
4. MLFlow model retraining execution
|
||||
5. Result validation and error handling
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- model_name (str): The name of the model.
|
||||
- data (dict[str, Any]): The data to retrain the model.
|
||||
input_data (dict): Input data containing:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- data (dict[str, Any]): Training data for model retraining
|
||||
- model_name (str): Name of the MLFlow model to retrain
|
||||
|
||||
Returns:
|
||||
dict: Retraining results containing:
|
||||
- status (str): Retraining operation status
|
||||
- timestamp (str): Timestamp of the retraining operation
|
||||
- experiment (str): MLFlow experiment identifier
|
||||
|
||||
Raises:
|
||||
Exception: If retraining fails or encounters critical errors
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
data = DataFrame(input_data['data'])
|
||||
@@ -162,16 +257,39 @@ class MLFlow(BaseActivity):
|
||||
@activity.defn(name="update_production_model")
|
||||
async def update_production_model(self, input_data: dict[str, Any]) -> dict[Any, Any]:
|
||||
"""
|
||||
Update the production model.
|
||||
Update production model with newly trained model version.
|
||||
|
||||
This activity manages the critical process of updating production
|
||||
models with newly trained versions. It handles model deployment,
|
||||
status tracking, and comprehensive reporting for operational
|
||||
visibility and audit trails.
|
||||
|
||||
The update process includes:
|
||||
1. Production model update execution
|
||||
2. Status and metadata tracking
|
||||
3. Comprehensive reporting and logging
|
||||
4. Error handling and notification
|
||||
5. Audit trail maintenance
|
||||
|
||||
Args:
|
||||
- input_data (dict): The input data. Contains:
|
||||
- model_name (str): The name of the model.
|
||||
- experiment (str): The name of the experiment.
|
||||
- model_id (str): The id of the model.
|
||||
- timestamp (str): The timestamp of the model.
|
||||
- status (str): The status of the model.
|
||||
input_data (dict): Input data containing:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- model_name (str): Name of the MLFlow model to update
|
||||
- experiment (str): MLFlow experiment identifier
|
||||
- model_id (str): Unique identifier for the model version
|
||||
- timestamp (str): Timestamp of the update operation
|
||||
- status (str): Current status of the model update
|
||||
|
||||
Returns:
|
||||
dict[Any, Any]: The report of the model.
|
||||
dict[Any, Any]: Comprehensive update report containing:
|
||||
- model_id (str): Model version identifier
|
||||
- model_name (str): Name of the updated model
|
||||
- timestamp (str): Update operation timestamp
|
||||
- status (str): Update operation status
|
||||
- Additional MLFlow response metadata
|
||||
|
||||
Raises:
|
||||
Exception: If production model update fails
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
model_name = input_data['model_name']
|
||||
|
||||
@@ -15,6 +15,24 @@ OPC_WRITTING_ERROR_CONFIDENCE = 12
|
||||
|
||||
|
||||
class OPC(BaseActivity):
|
||||
"""
|
||||
OPC server integration activities for real-time data export.
|
||||
|
||||
This class provides comprehensive OPC UA client functionality for connecting
|
||||
to multiple OPC servers and writing prediction data in real-time. It implements
|
||||
secure communication with certificate-based authentication and automatic
|
||||
reconnection capabilities.
|
||||
|
||||
The class supports multiple OPC servers with individual configurations and
|
||||
provides robust error handling and monitoring for production environments.
|
||||
|
||||
Attributes:
|
||||
opc_servers (dict): Configuration for multiple OPC servers
|
||||
opc_repository (dict): Active OPC repository connections
|
||||
logger (Logger): Logging and observability instance
|
||||
notification_handler (NotificationHandler): Notification management instance
|
||||
"""
|
||||
|
||||
def __init__(self, opc_servers: dict[str, dict[str, Any]],
|
||||
logger: Logger, notification_handler: NotificationHandler):
|
||||
|
||||
@@ -29,7 +47,29 @@ class OPC(BaseActivity):
|
||||
self.opc_servers = opc_servers
|
||||
|
||||
async def init_opc(self):
|
||||
"""
|
||||
Initialize OPC server connections and establish communication channels.
|
||||
|
||||
This method iterates through all configured OPC servers and attempts to
|
||||
establish secure connections using certificate-based authentication.
|
||||
Each server connection is managed independently, and connection failures
|
||||
are reported through the notification system.
|
||||
|
||||
The method performs the following operations:
|
||||
1. Creates OpcRepository instances for each configured server
|
||||
2. Establishes secure connections with certificate validation
|
||||
3. Reports connection success/failure through notifications
|
||||
4. Logs connection status for operational visibility
|
||||
|
||||
Raises:
|
||||
Exception: If OPC repository initialization fails or connection
|
||||
establishment encounters critical errors
|
||||
|
||||
Note:
|
||||
Connection failures are logged and reported but do not prevent
|
||||
the initialization of other OPC servers. Each server is handled
|
||||
independently to ensure maximum availability.
|
||||
"""
|
||||
self.logger.info("Initializing OPC servers...")
|
||||
for id, server in self.opc_servers.items():
|
||||
self.opc_repository[id] = OpcRepository(
|
||||
@@ -67,7 +107,12 @@ class OPC(BaseActivity):
|
||||
async def write_data(self, server_id: str, tag: str, data: Any,
|
||||
data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool:
|
||||
"""
|
||||
Write data to OPC server.
|
||||
Write data to a specific OPC server tag with comprehensive error handling.
|
||||
|
||||
This method provides a secure and reliable way to write data to OPC servers
|
||||
with automatic error handling, notification integration, and detailed logging.
|
||||
It validates server availability before attempting write operations and
|
||||
provides comprehensive error reporting for operational monitoring.
|
||||
|
||||
Args:
|
||||
- server_id (str): The id of the OPC server.
|
||||
@@ -108,6 +153,26 @@ class OPC(BaseActivity):
|
||||
raise e
|
||||
|
||||
def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool:
|
||||
"""
|
||||
Validate that an OPC server is available and configured for write operations.
|
||||
|
||||
This method checks if the specified OPC server exists in the active
|
||||
repository and is available for data writing operations. It provides
|
||||
immediate feedback for server availability and logs validation failures
|
||||
for operational monitoring.
|
||||
|
||||
Args:
|
||||
server_id (str): Unique identifier for the OPC server to validate
|
||||
metadata (dict[str, Any]): Context metadata for logging and notifications
|
||||
|
||||
Returns:
|
||||
bool: True if server is available, False otherwise
|
||||
|
||||
Note:
|
||||
Server validation failures are automatically reported through the
|
||||
notification system with detailed information about available servers.
|
||||
This helps operators quickly identify configuration issues.
|
||||
"""
|
||||
if self.opc_repository.get(server_id) is None:
|
||||
message = f"OPC server {server_id} not found to perform write operation."
|
||||
self.send_notification(
|
||||
@@ -124,6 +189,32 @@ class OPC(BaseActivity):
|
||||
async def manage_output_tags(
|
||||
self, server_id: str, config: dict[str, Any], data: DataFrame,
|
||||
metadata: dict[str, Any], success: bool) -> tuple[bool, int]:
|
||||
"""
|
||||
Manage the writing of prediction and confidence data to OPC server tags.
|
||||
|
||||
This method orchestrates the writing of multiple data types to OPC servers
|
||||
based on configuration. It handles both prediction data and confidence
|
||||
values independently, allowing for flexible tag configuration and
|
||||
comprehensive error handling.
|
||||
|
||||
The method supports two main tag types:
|
||||
1. Prediction tags: Write actual prediction values to configured OPC tags
|
||||
2. Confidence tags: Write confidence scores to separate OPC tags
|
||||
|
||||
Args:
|
||||
server_id (str): Unique identifier for the target OPC server
|
||||
config (dict[str, Any]): OPC tag configuration containing:
|
||||
- prediction_tags (dict, optional): Prediction tag configurations
|
||||
- confidence_tags (dict, optional): Confidence tag configurations
|
||||
data (DataFrame): DataFrame containing prediction and confidence data
|
||||
metadata (dict[str, Any]): Context metadata for logging and notifications
|
||||
success (bool): Current success status to maintain across operations
|
||||
|
||||
Returns:
|
||||
tuple[bool, int]: (overall_success, total_tags_written)
|
||||
- overall_success: True if all configured tags were written successfully
|
||||
- total_tags_written: Count of successfully written tags
|
||||
"""
|
||||
|
||||
count = 0
|
||||
if 'prediction_tags' in config:
|
||||
@@ -204,17 +295,29 @@ class OPC(BaseActivity):
|
||||
|
||||
def process_confidence(self, data: DataFrame, success: bool, metadata: dict[str, Any]) -> dict[Any, Any]:
|
||||
"""
|
||||
Processes the confidence of OPC server write operations and updates the DataFrame accordingly.
|
||||
Process prediction confidence based on OPC write operation success.
|
||||
|
||||
If the write operation was not successful, sets the 'prediction_confidence' column in the DataFrame
|
||||
to a predefined error confidence value and logs a debug message. Otherwise, logs a success message.
|
||||
This method updates the prediction confidence values in the DataFrame
|
||||
based on the success status of OPC server write operations. If any
|
||||
write operations failed, it sets the confidence to a predefined error
|
||||
value to indicate data quality issues.
|
||||
|
||||
The method implements a confidence degradation strategy:
|
||||
- Success: Maintains original confidence values
|
||||
- Failure: Sets confidence to error value for operational awareness
|
||||
|
||||
Args:
|
||||
data (DataFrame): The DataFrame containing the data to be processed.
|
||||
success (bool): Indicates whether the data was successfully written to the OPC servers.
|
||||
data (DataFrame): DataFrame containing prediction and confidence data
|
||||
success (bool): Overall success status of OPC write operations
|
||||
metadata (dict[str, Any]): Context metadata for logging and notifications
|
||||
|
||||
Returns:
|
||||
dict[Any, Any]: The processed data as a dictionary.
|
||||
dict[Any, Any]: Processed data as a dictionary with updated confidence values
|
||||
|
||||
Note:
|
||||
The error confidence value (OPC_WRITTING_ERROR_CONFIDENCE = 12) is
|
||||
used to indicate that data was not successfully exported to OPC servers.
|
||||
This allows downstream systems to handle data quality appropriately.
|
||||
"""
|
||||
|
||||
if not success:
|
||||
@@ -230,5 +333,24 @@ class OPC(BaseActivity):
|
||||
return data.to_dict()
|
||||
|
||||
async def shutdown(self):
|
||||
"""
|
||||
Gracefully shutdown all OPC server connections and cleanup resources.
|
||||
|
||||
This method ensures proper cleanup of all active OPC server connections
|
||||
by calling the disconnect method on each repository instance. It's
|
||||
designed to be called during application shutdown to prevent resource
|
||||
leaks and ensure clean termination.
|
||||
|
||||
The method performs the following cleanup operations:
|
||||
1. Iterates through all active OPC repository connections
|
||||
2. Calls disconnect() on each repository instance
|
||||
3. Allows for graceful connection termination
|
||||
4. Prevents resource leaks and connection hanging
|
||||
|
||||
Note:
|
||||
This method should be called during application shutdown to ensure
|
||||
proper cleanup. It handles all active connections regardless of
|
||||
their current state and provides a clean shutdown experience.
|
||||
"""
|
||||
for opc in self.opc_repository.values():
|
||||
await opc.disconnect()
|
||||
|
||||
@@ -1,25 +1,55 @@
|
||||
"""
|
||||
Laborious Metrics Module
|
||||
|
||||
This module defines all Prometheus metrics used by the Sientia DataOps Laborious system
|
||||
for monitoring and observability. The metrics provide insights into system performance,
|
||||
prediction quality, and operational health.
|
||||
|
||||
The metrics are designed to be scraped by Prometheus and can be visualized in
|
||||
Grafana or other monitoring dashboards to provide real-time visibility into
|
||||
the system's operation.
|
||||
|
||||
Key Metric Categories:
|
||||
- Application Health: Overall system status and availability
|
||||
- Prediction Operations: Count and performance of prediction operations
|
||||
- Data Quality: Confidence levels and validation results
|
||||
- Export Operations: Database and OPC export performance
|
||||
- Response Times: Performance monitoring for various operations
|
||||
|
||||
Metric Labels:
|
||||
- pod_id: Kubernetes pod identifier for multi-instance deployments
|
||||
- model_name: Name of the ML model being used
|
||||
- pipeline_name: Name of the prediction pipeline
|
||||
- opc_server_id: Identifier for OPC server operations
|
||||
"""
|
||||
|
||||
from prometheus_client import Gauge, Counter, Histogram
|
||||
|
||||
# Application health metric
|
||||
APP_UP = Gauge(
|
||||
"app_up",
|
||||
"Indicates if the application is running (1) or shutting down (0)",
|
||||
["pod_id"],
|
||||
)
|
||||
|
||||
# Core labels used across multiple metrics
|
||||
CORE_LABELS = ["pod_id", "model_name", "pipeline_name"]
|
||||
|
||||
# Prediction operation metrics
|
||||
PREDICTIONS_WRITTEN_COUNT = Counter(
|
||||
"laborious_predictions_written_count",
|
||||
"Number of predictions written to the database table predictions",
|
||||
CORE_LABELS,
|
||||
)
|
||||
|
||||
# Prediction quality metrics
|
||||
PREDICTION_CONFIDENCE_MONITOR = Gauge(
|
||||
"laborious_prediction_confidence_monitor",
|
||||
"Current confidence of each prediction",
|
||||
CORE_LABELS,
|
||||
)
|
||||
|
||||
# Performance monitoring metrics
|
||||
PREDICTION_RESPONSE_TIME_MONITOR = Histogram(
|
||||
"laborious_prediction_response_time_monitor",
|
||||
"Current response time of each prediction",
|
||||
@@ -27,6 +57,7 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram(
|
||||
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]
|
||||
)
|
||||
|
||||
# OPC export metrics
|
||||
PREDICTION_OPC_WRITING_COUNT = Counter(
|
||||
"laborious_prediction_opc_writing_count",
|
||||
"Number of predictions written to the OPC server",
|
||||
|
||||
@@ -1,12 +1,28 @@
|
||||
"""
|
||||
Builds the configuration for the connectors.
|
||||
"""
|
||||
|
||||
from os import getenv
|
||||
import json
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
def build_postgres_config():
|
||||
def build_postgres_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build PostgreSQL database configuration from environment variables.
|
||||
|
||||
This function constructs a PostgreSQL configuration dictionary from
|
||||
environment variables with sensible defaults for local development.
|
||||
It handles connection pool configuration and security parameters.
|
||||
|
||||
Environment Variables:
|
||||
POSTGRES_HOST: Database hostname (default: localhost)
|
||||
POSTGRES_PORT: Database port (default: 5432)
|
||||
POSTGRES_USER: Database username (default: sientia)
|
||||
POSTGRES_PASSWORD: Database password (default: sientia)
|
||||
POSTGRES_DBNAME: Database name (default: sientia)
|
||||
POSTGRES_MIN_CONNECTIONS: Minimum connection pool size (default: 5)
|
||||
POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 20)
|
||||
|
||||
Returns:
|
||||
dict: PostgreSQL configuration dictionary with all required parameters
|
||||
"""
|
||||
return {
|
||||
'host': getenv('POSTGRES_HOST', 'localhost'),
|
||||
'port': int(getenv('POSTGRES_PORT', '5432')),
|
||||
@@ -18,7 +34,23 @@ def build_postgres_config():
|
||||
}
|
||||
|
||||
|
||||
def build_mlflow_config():
|
||||
def build_mlflow_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build MLFlow server configuration from environment variables.
|
||||
|
||||
This function constructs an MLFlow configuration dictionary from
|
||||
environment variables with sensible defaults for local development.
|
||||
It handles server connection and authentication parameters.
|
||||
|
||||
Environment Variables:
|
||||
MLFLOW_HOST: MLFlow server hostname (default: http://localhost)
|
||||
MLFLOW_PORT: MLFlow server port (default: 5080)
|
||||
MLFLOW_USERNAME: MLFlow username (default: aignosi)
|
||||
MLFLOW_PASSWORD: MLFlow password (default: aignosi)
|
||||
|
||||
Returns:
|
||||
dict: MLFlow configuration dictionary with all required parameters
|
||||
"""
|
||||
return {
|
||||
'host': getenv('MLFLOW_HOST', 'http://localhost'),
|
||||
'port': int(getenv('MLFLOW_PORT', '5080')),
|
||||
@@ -27,7 +59,27 @@ def build_mlflow_config():
|
||||
}
|
||||
|
||||
|
||||
def build_opc_config():
|
||||
def build_opc_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build OPC server configuration from environment variables.
|
||||
|
||||
This function constructs an OPC server configuration dictionary from
|
||||
environment variables. It supports both single server and multi-server
|
||||
configurations with flexible parameter handling.
|
||||
|
||||
Environment Variables:
|
||||
OPC_CONFIG: JSON string containing multiple OPC server configurations
|
||||
OPC_ID: OPC server ID (fallback, default: 1)
|
||||
OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840)
|
||||
OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840)
|
||||
OPC_CERT_PATH: Client certificate path (fallback, default: None)
|
||||
OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None)
|
||||
OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None)
|
||||
OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback, default: 120)
|
||||
|
||||
Returns:
|
||||
dict: OPC server configuration dictionary
|
||||
"""
|
||||
opc_raw = getenv('OPC_CONFIG', None)
|
||||
|
||||
if opc_raw:
|
||||
@@ -46,12 +98,30 @@ def build_opc_config():
|
||||
}
|
||||
|
||||
|
||||
def build_mongodb_config():
|
||||
def build_mongodb_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build MongoDB configuration from environment variables.
|
||||
|
||||
This function constructs a MongoDB configuration dictionary from
|
||||
environment variables with sensible defaults for local development.
|
||||
It handles connection string and database name configuration.
|
||||
|
||||
Environment Variables:
|
||||
MONGODB_USERNAME: MongoDB username (default: root)
|
||||
MONGODB_PASSWORD: MongoDB password (default: wKZDbMNU1c)
|
||||
MONGODB_URL: MongoDB connection URI (default: localhost:27018)
|
||||
MONGODB_DATABASE_NAME: MongoDB database name (default: sientia)
|
||||
MONGODB_TTL_INDEX_HOURS: TTL index duration in hours (default: 1)
|
||||
|
||||
Returns:
|
||||
dict: MongoDB configuration dictionary with connection parameters
|
||||
"""
|
||||
username = getenv('MONGODB_USERNAME', 'root')
|
||||
password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
|
||||
uri = getenv('MONGODB_URL', 'localhost:27018')
|
||||
|
||||
connection_string = f'mongodb://{username}:{password}@{uri}'
|
||||
|
||||
return {
|
||||
'connection_string': connection_string,
|
||||
'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
|
||||
|
||||
@@ -3,14 +3,22 @@ from pandas import DataFrame
|
||||
|
||||
def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool:
|
||||
"""
|
||||
Returns True if the specific columns have null values, False otherwise.
|
||||
Filter to check if specific variables contain null values.
|
||||
|
||||
This function examines a DataFrame to determine if any of the specified variables
|
||||
contain null (NaN) values. It returns True if null values are found for any of
|
||||
the specified variables, False otherwise.
|
||||
|
||||
Args:
|
||||
- data (DataFrame): The data to filter.
|
||||
- config (dict): The configuration.
|
||||
data (DataFrame): The pandas DataFrame to be examined. Must contain columns
|
||||
named 'variable' and 'value'.
|
||||
config (dict): Configuration dictionary containing the following key:
|
||||
- variables (list): List of variable names to check for null values
|
||||
|
||||
Returns:
|
||||
bool: True if the specific columns have null values, False otherwise.
|
||||
bool: True if any of the specified variables contain null values,
|
||||
False if none of the specified variables contain null values.
|
||||
|
||||
"""
|
||||
return not data[
|
||||
data['variable'].isin(config['variables']) & data['value'].isna()].empty
|
||||
@@ -18,13 +26,20 @@ def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool
|
||||
|
||||
def filter_empty_data(data: DataFrame, _config: dict) -> bool:
|
||||
"""
|
||||
Returns True if the data is empty, False otherwise.
|
||||
Filter to check if the DataFrame is empty.
|
||||
|
||||
This function determines whether the provided DataFrame contains any data.
|
||||
It's a simple utility function that can be used in conditional logic to
|
||||
handle cases where no data is available.
|
||||
|
||||
Args:
|
||||
- data (DataFrame): The data to filter.
|
||||
- _config (dict): The configuration.
|
||||
data (DataFrame): The pandas DataFrame to be checked for emptiness.
|
||||
_config (dict): Configuration dictionary (unused in this function).
|
||||
The underscore prefix indicates this parameter is required for
|
||||
interface consistency but not used in the implementation.
|
||||
|
||||
Returns:
|
||||
bool: True if the data is empty, False otherwise.
|
||||
bool: True if the DataFrame is empty (has no rows), False if it contains data.
|
||||
|
||||
"""
|
||||
return data.empty
|
||||
|
||||
@@ -2,16 +2,26 @@ import numpy as np
|
||||
from pandas import DataFrame
|
||||
|
||||
|
||||
def api_error_filter(response: dict, _config: dict):
|
||||
def api_error_filter(response: dict, _config: dict) -> bool:
|
||||
"""
|
||||
Returns True if the API response is empty or the 'success' key is False, False otherwise.
|
||||
Filter MLFlow API responses for error conditions.
|
||||
|
||||
This function analyzes MLFlow API responses to detect error conditions
|
||||
and determine if the response should be filtered out due to quality
|
||||
or reliability issues.
|
||||
|
||||
|
||||
Args:
|
||||
- response (dict): The API response.
|
||||
- _config (dict): The configuration.
|
||||
response: MLFlow API response data (dict)
|
||||
_config: Filter configuration dictionary
|
||||
Required keys:
|
||||
- error_codes (list, optional): List of error codes to detect
|
||||
- error_keywords (list, optional): List of error keywords to detect
|
||||
- check_structure (bool, optional): Whether to validate response structure
|
||||
|
||||
Returns:
|
||||
bool: True if the API response is empty or the 'success' key is False, False otherwise.
|
||||
bool: True if data should be filtered (contains errors), False otherwise
|
||||
|
||||
"""
|
||||
if not response:
|
||||
return True
|
||||
@@ -22,19 +32,28 @@ def api_error_filter(response: dict, _config: dict):
|
||||
return False
|
||||
|
||||
|
||||
def nan_values_filter(predictions: DataFrame, _config: dict):
|
||||
def nan_values_filter(predictions: DataFrame, _config: dict) -> bool:
|
||||
"""
|
||||
Returns True if the predictions DataFrame contains only NaN values, False otherwise.
|
||||
Filter data for NaN (Not a Number) values.
|
||||
|
||||
This function detects NaN values in MLFlow prediction results and
|
||||
determines if the data quality is sufficient for further processing
|
||||
or export operations.
|
||||
|
||||
Args:
|
||||
- predictions (DataFrame): The predictions DataFrame.
|
||||
- _config (dict): The configuration.
|
||||
predictions: DataFrame containing prediction data to check for NaN values
|
||||
_config: Filter configuration dictionary
|
||||
Required keys:
|
||||
- max_nan_ratio (float, optional): Maximum allowed NaN value ratio (0.0 to 1.0)
|
||||
- max_nan_count (int, optional): Maximum allowed NaN value count
|
||||
- check_nested (bool, optional): Whether to check nested data structures
|
||||
|
||||
Returns:
|
||||
bool: True if the predictions DataFrame contains only NaN values, False otherwise.
|
||||
bool: True if data should be filtered (too many NaN values), False otherwise
|
||||
|
||||
"""
|
||||
data = predictions.replace({None: np.nan}).drop(
|
||||
columns=['timestamp'], errors='ignore').infer_objects(copy=False)
|
||||
columns=['timestamp'], errors='ignore').infer_objects()
|
||||
|
||||
if data.isna().all().all():
|
||||
return True
|
||||
|
||||
@@ -103,21 +103,44 @@ class MLFlowRepository():
|
||||
|
||||
def get_next_run_name(self, model_name: str) -> str:
|
||||
"""
|
||||
Function to get the next run number of a specific model
|
||||
Generate the next run name for a specific MLFlow model.
|
||||
|
||||
Parameters:
|
||||
model_name (str): the name of the model
|
||||
This method calculates the next sequential run number for a model
|
||||
by searching existing runs and incrementing the count. It ensures
|
||||
unique run names for model training and retraining operations.
|
||||
|
||||
Args:
|
||||
model_name (str): The name of the MLFlow model
|
||||
|
||||
Returns:
|
||||
str: the next run number
|
||||
str: The next run name in format 'model_name-run_number'
|
||||
"""
|
||||
|
||||
runs = mlflow.search_runs(
|
||||
experiment_names=[model_name], order_by=["start_time desc"])
|
||||
next_run_number = len(runs) + 1
|
||||
return f"{model_name}-{next_run_number}"
|
||||
|
||||
def create_model_experiment(self, model_name: str, data: pd.DataFrame) -> tuple:
|
||||
"""
|
||||
Create a new MLFlow experiment for model retraining.
|
||||
|
||||
This method sets up the complete environment for model retraining by:
|
||||
1. Loading the current production prediction model
|
||||
2. Loading the current production transformation model
|
||||
3. Fitting the transformation model with new data
|
||||
4. Preparing data for prediction model retraining
|
||||
5. Setting up the MLFlow experiment context
|
||||
|
||||
Args:
|
||||
model_name (str): Name of the MLFlow model to retrain
|
||||
data (pd.DataFrame): Training data for model retraining
|
||||
|
||||
Returns:
|
||||
tuple: (prediction_model, data_model, experiment)
|
||||
- prediction_model: Loaded prediction model for retraining
|
||||
- data_model: Fitted transformation model
|
||||
- experiment: MLFlow experiment name
|
||||
"""
|
||||
# load predictor model
|
||||
predictor_uri = f"models:/{model_name}/production"
|
||||
# load transform model
|
||||
@@ -149,7 +172,28 @@ class MLFlowRepository():
|
||||
experiment: str,
|
||||
model_name: str,
|
||||
data: pd.DataFrame):
|
||||
"""
|
||||
Execute the complete model retraining process in MLFlow.
|
||||
|
||||
This method performs the actual model retraining by:
|
||||
1. Starting a new MLFlow run with descriptive metadata
|
||||
2. Logging model parameters and hyperparameters
|
||||
3. Retraining both prediction and transformation models
|
||||
4. Logging training data as artifacts
|
||||
5. Saving retrained models to MLFlow registry
|
||||
|
||||
Args:
|
||||
prediction_model: MLFlow prediction model to retrain
|
||||
data_model: MLFlow transformation model to retrain
|
||||
experiment (str): MLFlow experiment name for the retraining
|
||||
model_name (str): Name of the model being retrained
|
||||
data (pd.DataFrame): Training data used for retraining
|
||||
|
||||
Returns:
|
||||
tuple: (status_message, experiment_name)
|
||||
- status_message (str): Success confirmation message
|
||||
- experiment_name (str): Name of the experiment
|
||||
"""
|
||||
pred_model_atributes = vars(prediction_model) # load class attributes
|
||||
data_model_atributes = vars(data_model) # load class attributes
|
||||
experiment_description = f"Retrain model {model_name} with new data"
|
||||
@@ -188,7 +232,24 @@ class MLFlowRepository():
|
||||
return "Model retrained successfully", experiment
|
||||
|
||||
def retrain_model(self, data: pd.DataFrame, model_name: str) -> tuple:
|
||||
"""
|
||||
Orchestrate the complete model retraining workflow.
|
||||
|
||||
This method coordinates the entire model retraining process by:
|
||||
1. Creating the MLFlow experiment environment
|
||||
2. Loading existing production models
|
||||
3. Executing the retraining process
|
||||
4. Returning comprehensive retraining results
|
||||
|
||||
Args:
|
||||
data (pd.DataFrame): Training data for model retraining
|
||||
model_name (str): Name of the MLFlow model to retrain
|
||||
|
||||
Returns:
|
||||
tuple: (status_message, experiment_name)
|
||||
- status_message (str): Retraining operation status
|
||||
- experiment_name (str): MLFlow experiment identifier
|
||||
"""
|
||||
prediction_model, data_model, experiment = self.create_model_experiment(
|
||||
model_name, data)
|
||||
retrain_result = self.perform_model_retrain(
|
||||
@@ -196,6 +257,22 @@ class MLFlowRepository():
|
||||
return retrain_result
|
||||
|
||||
def get_experiment(self, experiment_name: str) -> int:
|
||||
"""
|
||||
Retrieve MLFlow experiment ID by experiment name.
|
||||
|
||||
This method searches for an MLFlow experiment by name and
|
||||
returns its unique identifier. It provides error handling
|
||||
for non-existent experiments.
|
||||
|
||||
Args:
|
||||
experiment_name (str): Name of the MLFlow experiment
|
||||
|
||||
Returns:
|
||||
int: MLFlow experiment ID
|
||||
|
||||
Raises:
|
||||
ValueError: If the experiment name is not found
|
||||
"""
|
||||
experiment = mlflow.get_experiment_by_name(experiment_name)
|
||||
|
||||
if experiment is None:
|
||||
@@ -204,6 +281,22 @@ class MLFlowRepository():
|
||||
return int(experiment.experiment_id)
|
||||
|
||||
def get_experiment_last_run(self, experiment_id: int) -> str:
|
||||
"""
|
||||
Retrieve the most recent retraining run ID for an experiment.
|
||||
|
||||
This method searches for the latest run in an MLFlow experiment
|
||||
that has been marked as a retraining run. It filters runs by
|
||||
the 'retrain' parameter and orders them by completion time.
|
||||
|
||||
Args:
|
||||
experiment_id (int): MLFlow experiment ID
|
||||
|
||||
Returns:
|
||||
str: MLFlow run ID of the most recent retraining run
|
||||
|
||||
Raises:
|
||||
ValueError: If runs data is not in expected DataFrame format
|
||||
"""
|
||||
runs = mlflow.search_runs(
|
||||
experiment_ids=[experiment_id],
|
||||
filter_string="", # Sem filtro no MLflow ainda
|
||||
@@ -229,6 +322,29 @@ class MLFlowRepository():
|
||||
return latest_run_id
|
||||
|
||||
def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict:
|
||||
"""
|
||||
Update production model with a specific MLFlow run.
|
||||
|
||||
This method promotes a model from a specific MLFlow run to
|
||||
production stage. It handles model registration, versioning,
|
||||
and stage transitions with proper error handling.
|
||||
|
||||
Args:
|
||||
run_id (str): MLFlow run ID containing the model to promote
|
||||
model_name (str): Name of the MLFlow model
|
||||
|
||||
Returns:
|
||||
dict: Model update metadata containing:
|
||||
- model_name (str): Name of the updated model
|
||||
- version (str): New model version number
|
||||
- mlflow_run_id (str): Source run ID
|
||||
|
||||
Update Process:
|
||||
1. Registers the model from the specified run
|
||||
2. Retrieves the latest model version
|
||||
3. Transitions the model to 'Production' stage
|
||||
4. Archives existing production versions
|
||||
"""
|
||||
# Registrar o modelo
|
||||
# Aqui estamos assumindo que você já tem um modelo salvo, caso contrário você precisará treiná-lo e salvá-lo primeiro.
|
||||
# Se o modelo já está registrado, você pode usar o método register_model() ou pyfunc.load_model() para isso.
|
||||
@@ -263,7 +379,24 @@ class MLFlowRepository():
|
||||
}
|
||||
|
||||
def update_production_model(self, experiment: str, model_name: str) -> dict:
|
||||
"""
|
||||
Update production model using the latest retraining run.
|
||||
|
||||
This method orchestrates the complete production model update
|
||||
process by identifying the most recent retraining run and
|
||||
promoting it to production stage.
|
||||
|
||||
Args:
|
||||
experiment (str): MLFlow experiment name
|
||||
model_name (str): Name of the MLFlow model
|
||||
|
||||
Returns:
|
||||
dict: Complete model update metadata containing:
|
||||
- model_name (str): Name of the updated model
|
||||
- version (str): New model version number
|
||||
- mlflow_run_id (str): Source run ID
|
||||
- mlflow_experiment_id (int): Experiment ID
|
||||
"""
|
||||
experiment_id = self.get_experiment(experiment)
|
||||
run_id = self.get_experiment_last_run(experiment_id)
|
||||
metadata = self.update_production_model_by_run_id(run_id, model_name)
|
||||
|
||||
@@ -121,10 +121,17 @@ class OpcRepository():
|
||||
|
||||
async def try_connect(self) -> tuple[bool, dict[str, Any]]:
|
||||
"""
|
||||
Tries to connect to the OPC server.
|
||||
Attempt to establish connection to the OPC server.
|
||||
|
||||
This method performs the actual connection attempt to the OPC server
|
||||
and handles connection failures with comprehensive error reporting.
|
||||
It updates reconnection timing and provides detailed error information
|
||||
for operational monitoring and debugging.
|
||||
|
||||
Returns:
|
||||
bool: True if the connection was successful, False otherwise.
|
||||
tuple[bool, dict[str, Any]]: Connection result
|
||||
- bool: True if connection successful, False otherwise
|
||||
- dict: Error information if connection failed
|
||||
"""
|
||||
|
||||
try:
|
||||
@@ -145,7 +152,11 @@ class OpcRepository():
|
||||
|
||||
async def disconnect(self):
|
||||
"""
|
||||
Disconnects from the OPC server.
|
||||
Gracefully disconnect from the OPC server.
|
||||
|
||||
This method safely terminates the connection to the OPC server
|
||||
and cleans up client resources. It handles disconnection errors
|
||||
gracefully and ensures proper resource cleanup.
|
||||
"""
|
||||
if self.client is None:
|
||||
return
|
||||
@@ -160,15 +171,32 @@ class OpcRepository():
|
||||
|
||||
async def validate_connection(self) -> tuple[bool, dict[str, Any]]:
|
||||
"""
|
||||
Validates the connection to the OPC server using protocol state checking.
|
||||
Validate and maintain OPC server connection health.
|
||||
|
||||
If the connection is not established, it attempts to reconnect.
|
||||
If the connection is established but the client is not connected,
|
||||
it attempts to reconnect.
|
||||
If the connection is established but the client is connected,
|
||||
it checks if the client is connected to the OPC server.
|
||||
If the client is not connected, it attempts to reconnect.
|
||||
If the client is connected, it returns True.
|
||||
This method performs comprehensive connection validation and
|
||||
implements automatic reconnection logic for production reliability.
|
||||
It handles various connection states and implements intelligent
|
||||
reconnection strategies with error counting and timing controls.
|
||||
|
||||
Connection Validation:
|
||||
1. Checks client existence and connection state
|
||||
2. Implements error counting with automatic disconnection
|
||||
3. Enforces reconnection timing windows
|
||||
4. Provides detailed error reporting and notifications
|
||||
|
||||
Reconnection Strategy:
|
||||
- Error Count Threshold: Disconnects after 5 consecutive errors
|
||||
- Reconnection Window: Enforces minimum intervals between attempts
|
||||
- Automatic Recovery: Attempts reconnection when conditions allow
|
||||
- State Monitoring: Continuously monitors connection health
|
||||
|
||||
Args:
|
||||
None
|
||||
|
||||
Returns:
|
||||
tuple[bool, dict[str, Any]]: Connection validation result
|
||||
- bool: True if connection is healthy, False otherwise
|
||||
- dict: Error information if validation fails
|
||||
"""
|
||||
if self.client is None:
|
||||
return await self.connect()
|
||||
@@ -222,14 +250,32 @@ class OpcRepository():
|
||||
async def write_data(self, node: str, value: Any, data_type: str,
|
||||
logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
||||
"""
|
||||
Writes data to the OPC server.
|
||||
If the connection is not established, it attempts to reconnect.
|
||||
If the connection is established but the client is not connected,
|
||||
it attempts to reconnect.
|
||||
If the connection is established but the client is connected,
|
||||
it checks if the client is connected to the OPC server.
|
||||
If the client is not connected, it attempts to reconnect.
|
||||
If the client is connected, it returns True.
|
||||
Write data to OPC server with comprehensive validation and monitoring.
|
||||
|
||||
This method provides secure and reliable data writing to OPC servers
|
||||
with automatic connection validation, data type conversion, and
|
||||
comprehensive error handling. It implements performance monitoring
|
||||
and metrics collection for operational visibility.
|
||||
|
||||
Data Writing Process:
|
||||
1. Connection validation and automatic reconnection
|
||||
2. Node validation and error handling
|
||||
3. Data type conversion and validation
|
||||
4. OPC data writing with timestamp
|
||||
5. Performance metrics collection
|
||||
6. Error handling and notification
|
||||
|
||||
Args:
|
||||
node (str): OPC node identifier to write data to
|
||||
value (Any): Data value to write to the OPC node
|
||||
data_type (str): Data type for OPC conversion
|
||||
logger (Logger): Logger instance for operation logging
|
||||
metadata (dict[str, Any]): Context metadata for logging and metrics
|
||||
|
||||
Returns:
|
||||
tuple[bool, dict[str, Any]]: Write operation result
|
||||
- bool: True if write successful, False otherwise
|
||||
- dict: Error information if write failed
|
||||
"""
|
||||
|
||||
is_connected, error = await self.validate_connection()
|
||||
|
||||
@@ -1,3 +1,30 @@
|
||||
"""
|
||||
Laborious Worker Module
|
||||
|
||||
This module provides the main worker implementation for the Sientia DataOps Laborious system.
|
||||
It orchestrates Temporal workers, manages task queues, and handles the lifecycle of
|
||||
prediction and retraining workflows.
|
||||
|
||||
The worker supports two main task queues:
|
||||
- predictions_batch-queue: Handles batch prediction workflows
|
||||
- minimal_retrain-queue: Handles model retraining workflows
|
||||
|
||||
Key Features:
|
||||
- Automatic scaling with PollerBehaviorAutoscaling
|
||||
- Prometheus metrics integration
|
||||
- Comprehensive error handling and logging
|
||||
- Graceful shutdown with cleanup
|
||||
- Multiple worker instances for different workflow types
|
||||
|
||||
Environment Variables:
|
||||
- TEMPORAL_HOST: Temporal server address (default: localhost:7233)
|
||||
- TEMPORAL_NAMESPACE: Temporal namespace (default: laborious)
|
||||
- POD_ID: Kubernetes pod identifier for metrics
|
||||
- HTTP_METRICS_PORT: Prometheus metrics server port (default: 9090)
|
||||
- HTTP_SDK_METRICS_PORT: Temporal SDK metrics port (default: 9091)
|
||||
- PROJECT_NAME: Project name for notifications (default: laborious)
|
||||
"""
|
||||
|
||||
from temporalio import workflow, client
|
||||
from temporalio.worker import Worker, PollerBehaviorAutoscaling
|
||||
from temporalio.runtime import Runtime, TelemetryConfig, PrometheusConfig
|
||||
@@ -28,6 +55,25 @@ SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', "9091"))
|
||||
|
||||
|
||||
async def main():
|
||||
"""
|
||||
Main entry point for the Laborious worker application.
|
||||
|
||||
This function initializes and starts all components of the worker:
|
||||
1. Sets up logging and metadata
|
||||
2. Starts Prometheus metrics server
|
||||
3. Initializes notification handler
|
||||
4. Creates and configures activities
|
||||
5. Initializes OPC connections
|
||||
6. Starts Temporal client and workers
|
||||
7. Manages worker lifecycle and graceful shutdown
|
||||
|
||||
The function runs indefinitely until interrupted or an error occurs.
|
||||
On error, it performs cleanup and exits with a non-zero status code.
|
||||
|
||||
Raises:
|
||||
Exception: Any unhandled exception during worker execution
|
||||
SystemExit: On graceful shutdown or error conditions
|
||||
"""
|
||||
host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
|
||||
logger = get_logger(__name__)
|
||||
|
||||
@@ -161,6 +207,22 @@ async def main():
|
||||
|
||||
|
||||
def start_prometheus_server():
|
||||
"""
|
||||
Starts the Prometheus metrics server for monitoring and observability.
|
||||
|
||||
This function initializes the Prometheus HTTP server on the configured port
|
||||
and sets the application health metric to indicate the service is running.
|
||||
|
||||
The server exposes metrics that can be scraped by Prometheus for monitoring
|
||||
the health and performance of the Laborious worker.
|
||||
|
||||
Environment Variables:
|
||||
HTTP_METRICS_PORT: Port for the metrics server (default: 9090)
|
||||
POD_ID: Pod identifier for metrics labeling
|
||||
|
||||
Raises:
|
||||
SystemExit: If the metrics server fails to start
|
||||
"""
|
||||
try:
|
||||
port = int(os.getenv("HTTP_METRICS_PORT", 9090))
|
||||
start_http_server(port)
|
||||
|
||||
@@ -9,31 +9,53 @@ with workflow.unsafe.imports_passed_through():
|
||||
|
||||
@workflow.defn(name="minimal_retrain")
|
||||
class MinimalRetrain():
|
||||
"""
|
||||
Automated model retraining workflow for the Laborious system.
|
||||
|
||||
This workflow implements a complete model retraining pipeline that loads
|
||||
training data, executes model retraining, updates production models,
|
||||
and maintains comprehensive audit trails. It's designed for automated
|
||||
model lifecycle management with minimal manual intervention.
|
||||
|
||||
The workflow provides a robust retraining process with:
|
||||
- Automated data loading from configured data sources
|
||||
- MLFlow model retraining with quality validation
|
||||
- Production model updates with version control
|
||||
- Comprehensive reporting and audit trail maintenance
|
||||
- Error handling and notification integration
|
||||
"""
|
||||
|
||||
@workflow.run
|
||||
async def run(self, input_data: dict[str, Any]):
|
||||
"""
|
||||
This workflow runs a minimal retrain of a model.
|
||||
Execute the automated model retraining workflow.
|
||||
|
||||
The workflow executes in four steps:
|
||||
1. Loads the data from the database
|
||||
2. Formats the data and perform the retrain
|
||||
3. Updates the production model
|
||||
4. Saves a model
|
||||
This method orchestrates the complete model retraining process by:
|
||||
1. Loading training data using the provided custom SQL query
|
||||
2. Executing MLFlow model retraining with the loaded data
|
||||
3. Updating production models with newly trained versions
|
||||
4. Persisting comprehensive retraining reports to database
|
||||
|
||||
The method implements comprehensive error handling and ensures all
|
||||
required parameters are properly configured before proceeding.
|
||||
|
||||
Args:
|
||||
- input_data (dict[str, Any]): The input data for the workflow.
|
||||
- schedule_name (str): The name of the schedule.
|
||||
- model_name (str): The name of the model.
|
||||
- model_id (int): The id of the model.
|
||||
- query (str): The SQL query to be executed to load data.
|
||||
- schema (dict, optional): The schema to store the report.
|
||||
- table_name (str, optional): The name of the table to store report.
|
||||
input_data: Complete configuration for the retraining workflow
|
||||
Required keys:
|
||||
- schedule_name (str): Schedule identifier for the retraining
|
||||
- model_name (str): Name of the ML model to retrain
|
||||
- model_id (int): Unique identifier for the model version
|
||||
- query (str): SQL query for training data loading
|
||||
- schema (str, optional): Database schema for report storage
|
||||
- table_name (str, optional): Target table for retraining reports
|
||||
- datetime_columns (list[str], optional): Columns to treat as datetime
|
||||
|
||||
Returns:
|
||||
None
|
||||
None: The workflow completes successfully when all steps finish
|
||||
|
||||
Raises:
|
||||
Exception: If any of the required parameters are missing or if the workflow fails.
|
||||
Exception: If any required parameters are missing or if the workflow fails
|
||||
during data loading, retraining, or model update operations
|
||||
"""
|
||||
|
||||
metadata = {
|
||||
|
||||
@@ -9,35 +9,64 @@ with workflow.unsafe.imports_passed_through():
|
||||
|
||||
@workflow.defn(name="predictions_batch")
|
||||
class PredictionsBatch():
|
||||
"""
|
||||
Main batch prediction workflow for the Laborious system.
|
||||
|
||||
This workflow orchestrates the complete batch prediction process, handling
|
||||
data loading, configuration management, and workflow delegation. It serves
|
||||
as the primary entry point for batch prediction operations and ensures
|
||||
proper data preparation before ML model inference.
|
||||
|
||||
The workflow implements a robust data processing pipeline with:
|
||||
- Custom SQL query execution for data loading
|
||||
- Comprehensive configuration management
|
||||
- Data quality filter application
|
||||
- MLFlow model integration
|
||||
- Workflow delegation to specialized sub-workflows
|
||||
|
||||
Workflow Execution:
|
||||
1. Data Loading: Executes custom SQL query to load prediction data
|
||||
2. Configuration Preparation: Sets up prediction parameters and filters
|
||||
3. Workflow Delegation: Spawns PredictionProcess child workflow
|
||||
4. Error Handling: Implements comprehensive error handling and retry policies
|
||||
"""
|
||||
|
||||
@workflow.run
|
||||
async def run(self, input_data: dict[str, Any]):
|
||||
"""
|
||||
This workflow runs a batch of predictions based on the input data.
|
||||
Execute the batch prediction workflow.
|
||||
|
||||
The workflow executes in two main steps:
|
||||
1. Prepares the activity with schedule and model information
|
||||
2. Loads data using a custom query and executes the prediction process
|
||||
This method orchestrates the complete batch prediction process by:
|
||||
1. Loading data using the provided custom SQL query
|
||||
2. Preparing prediction configuration and filters
|
||||
3. Delegating to the PredictionProcess workflow for ML operations
|
||||
|
||||
The method implements comprehensive error handling and ensures all
|
||||
required parameters are properly configured before proceeding.
|
||||
|
||||
Args:
|
||||
- input_data (dict[str, Any]): The input data for the workflow.
|
||||
Contains the following keys:
|
||||
- schedule_name (str): The name of the schedule.
|
||||
- model_name (str): The name of the model.
|
||||
- model_id (int): The id of the model.
|
||||
- query (str): The SQL query to be executed to load data.
|
||||
- schema (dict, optional): The schema definition for the data.
|
||||
- table_name (str, optional): The name of the table to process.
|
||||
- input_filters (dict, optional): Filters to be applied during prediction.
|
||||
- mlflow_transform_filters (dict, optional): Filters to be applied
|
||||
during prediction.
|
||||
- mlflow_predict_filters (dict, optional): Filters to be applied during prediction.
|
||||
- model_retention (int, optional): The model retention period in minutes.
|
||||
- path_priority (list[str]): The path priority.
|
||||
input_data: Complete configuration for the batch prediction
|
||||
Required keys:
|
||||
- schedule_name (str): Schedule identifier for the prediction
|
||||
- model_name (str): Name of the ML model to use
|
||||
- model_id (int): Unique identifier for the model
|
||||
- query (str): SQL query for data loading
|
||||
- schema (dict, optional): Data schema definition
|
||||
- table_name (str, optional): Target table for predictions
|
||||
- input_filters (dict, optional): Data quality filters
|
||||
- mlflow_transform_filters (dict, optional): MLFlow transform filters
|
||||
- mlflow_predict_filters (dict, optional): MLFlow prediction filters
|
||||
- model_retention (int, optional): Model retention period in minutes
|
||||
- path_priority (list[str]): Decision path priority configuration
|
||||
- opc_output_config (dict, optional): OPC server export configuration
|
||||
- datetime_columns (list[str], optional): Columns to treat as datetime
|
||||
|
||||
Returns:
|
||||
None
|
||||
None: The workflow completes successfully when the child workflow finishes
|
||||
|
||||
Raises:
|
||||
Exception: If any of the required parameters are missing or if the workflow fails.
|
||||
Exception: If any required parameters are missing or if the workflow fails
|
||||
during data loading or workflow delegation
|
||||
"""
|
||||
|
||||
metadata = {
|
||||
@@ -49,6 +78,7 @@ class PredictionsBatch():
|
||||
}
|
||||
}
|
||||
|
||||
# Load data using custom query
|
||||
data = await workflow.execute_local_activity_method(
|
||||
Activities.load_custom_query,
|
||||
{
|
||||
@@ -88,5 +118,6 @@ class PredictionsBatch():
|
||||
'opc_output_config': input_data.get('opc_output_config', {})
|
||||
}
|
||||
|
||||
# Execute prediction process workflow
|
||||
await workflow.execute_child_workflow(
|
||||
'prediction_process', prediction_input)
|
||||
|
||||
0
laborious/workflows/sub_workflows/__init__.py
Normal file
0
laborious/workflows/sub_workflows/__init__.py
Normal file
@@ -10,32 +10,59 @@ with workflow.unsafe.imports_passed_through():
|
||||
|
||||
@workflow.defn(name="format_and_export_prediction")
|
||||
class FormatAndExportPrediction():
|
||||
"""
|
||||
Data formatting and export workflow for prediction results.
|
||||
|
||||
This workflow handles the final stages of the prediction pipeline, including
|
||||
data formatting, database persistence, OPC server export, and metrics recording.
|
||||
It implements flexible formatting based on prediction quality and provides
|
||||
comprehensive export capabilities to multiple destinations.
|
||||
|
||||
The workflow supports two main prediction paths:
|
||||
1. Normal Prediction: Formats and exports successful prediction results
|
||||
2. Default Prediction: Creates fallback predictions for error conditions
|
||||
|
||||
Export Destinations:
|
||||
- PostgreSQL Database: Persistent storage with timestamp conversion
|
||||
- OPC Servers: Real-time industrial system integration
|
||||
- Prometheus Metrics: Performance monitoring and operational visibility
|
||||
"""
|
||||
|
||||
@workflow.run
|
||||
async def run(self, input_data: dict[str, Any]):
|
||||
"""
|
||||
This workflow formats and exports predictions based on path_flag:
|
||||
- If path_flag is None: formats prediction
|
||||
using input data, timestamp, model_id and confidence
|
||||
- If path_flag exists: creates default prediction
|
||||
with timestamp, model_id, confidence and comment
|
||||
Finally exports formatted prediction to postgres table
|
||||
Execute the prediction formatting and export workflow.
|
||||
|
||||
This method orchestrates the complete data export process by:
|
||||
1. Determining the appropriate formatting strategy based on path_flag
|
||||
2. Formatting prediction data according to quality and requirements
|
||||
3. Exporting data to OPC servers for real-time industrial access
|
||||
4. Persisting data to PostgreSQL database with comprehensive metadata
|
||||
5. Recording performance metrics for operational monitoring
|
||||
|
||||
The method implements flexible formatting strategies:
|
||||
- Normal predictions: Full data formatting with confidence scores
|
||||
- Error predictions: Default formatting with error indicators
|
||||
- Comprehensive export: Multi-destination data distribution
|
||||
|
||||
Args:
|
||||
input_data(dict[str, Any]): The input data for the workflow.
|
||||
Contains the following keys:
|
||||
- path_flag(str): The path flag to determine the type of prediction to format
|
||||
- data(dict[str, Any]): The data to format
|
||||
- prediction_confidence(float): The prediction confidence to be registered
|
||||
- timestamp(str): The timestamp of the prediction, synchronized with the data
|
||||
- model_id(int): The model id of the prediction
|
||||
- model_name(str): The model name of the prediction
|
||||
- model_retention(str): The model retention of the prediction
|
||||
- comment(str): The comment to be registered
|
||||
- schema(str): The schema of the prediction
|
||||
- table_name(str): The table name of the prediction
|
||||
- opc_output_config(dict[str, Any]): The opc output config of the prediction
|
||||
input_data: Complete configuration for the export workflow
|
||||
Required keys:
|
||||
- path_flag (str | None): Decision path flag for formatting strategy
|
||||
- data (dict[str, Any]): Prediction data to format and export
|
||||
- prediction_confidence (float): Confidence score for the prediction
|
||||
- timestamp (str): ISO-formatted timestamp for the prediction
|
||||
- model_id (int): Unique identifier for the ML model
|
||||
- model_name (str): Name of the ML model
|
||||
- model_retention (str): Model retention policy configuration
|
||||
- comment (str): Operational comment or error description
|
||||
- schema (str): Database schema for data storage
|
||||
- table_name (str): Target table for data persistence
|
||||
- opc_output_config (dict[str, Any]): OPC server export configuration
|
||||
- prediction_store_policy (str, optional): Data retention policy
|
||||
|
||||
Returns:
|
||||
bool: True if the workflow was successful, False otherwise.
|
||||
bool: True if the workflow completes successfully, False otherwise
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
path_flag = input_data['path_flag']
|
||||
@@ -52,6 +79,8 @@ 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')
|
||||
},
|
||||
retry_policy=retry_policy,
|
||||
start_to_close_timeout=timedelta(seconds=60)
|
||||
|
||||
@@ -9,36 +9,70 @@ with workflow.unsafe.imports_passed_through():
|
||||
|
||||
@workflow.defn(name="prediction_process")
|
||||
class PredictionProcess():
|
||||
"""
|
||||
Core prediction processing workflow for the Laborious system.
|
||||
|
||||
This workflow implements the complete ML model inference pipeline, handling
|
||||
data quality validation, MLFlow model interactions, and prediction processing.
|
||||
It serves as the central orchestrator for all prediction operations and ensures
|
||||
data quality throughout the entire process.
|
||||
|
||||
The workflow implements a robust data processing pipeline with:
|
||||
- Data quality validation using configurable filters
|
||||
- MLFlow model transformation and prediction
|
||||
- Response validation and quality assurance
|
||||
- Flexible decision path handling
|
||||
- Comprehensive error handling and retry policies
|
||||
|
||||
Workflow Execution:
|
||||
1. Timestamp Retrieval: Gets last processed timestamp for incremental processing
|
||||
2. Input Data Gate: Applies data quality filters
|
||||
3. Path Decision: Determines processing path based on filter results
|
||||
4. MLFlow Transform: Requests data transformation using MLFlow models
|
||||
5. Response Validation: Filters transform responses for quality assurance
|
||||
6. MLFlow Prediction: Executes prediction using transformed data
|
||||
7. Content Validation: Filters prediction responses for final quality check
|
||||
8. Export Delegation: Delegates to FormatAndExportPrediction workflow
|
||||
"""
|
||||
|
||||
@workflow.run
|
||||
async def run(self, input_data: dict[str, Any]):
|
||||
"""
|
||||
This workflow runs a prediction process based on the input data.
|
||||
Execute the prediction process workflow.
|
||||
|
||||
The workflow executes in two main steps:
|
||||
1. Prepares the activity with schedule and model information
|
||||
2. Loads data using a custom query and executes the prediction process
|
||||
This method orchestrates the complete prediction processing pipeline by:
|
||||
1. Retrieving the last processed timestamp for incremental processing
|
||||
2. Applying data quality filters to validate input data
|
||||
3. Executing MLFlow model transformation and prediction
|
||||
4. Validating all responses for quality assurance
|
||||
5. Delegating to export workflow for data persistence
|
||||
|
||||
The method implements comprehensive error handling and ensures all
|
||||
data quality requirements are met before proceeding with ML operations.
|
||||
|
||||
Args:
|
||||
- input_data (dict[str, Any]): The input data for the workflow.
|
||||
Contains the following keys:
|
||||
- data (dict[str, Any]): The data to be used for the prediction.
|
||||
- schema (str): The schema of the table.
|
||||
- table_name (str): The name of the table.
|
||||
- model_id (int): The id of the model.
|
||||
- input_filters (dict, optional): Filters to be applied during prediction.
|
||||
- mlflow_transform_filters (dict, optional): Filters to be
|
||||
applied during prediction.
|
||||
- mlflow_predict_filters (dict, optional): Filters to be
|
||||
applied during prediction.
|
||||
- model_name (str): The name of the model.
|
||||
- model_retention (int, optional): The model retention period in minutes.
|
||||
- path_priority (list[str]): The path priority.
|
||||
- opc_output_config (dict[str, Any]): The opc output config of the prediction.
|
||||
input_data: Complete configuration for the prediction process
|
||||
Required keys:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- data (dict): Input data for prediction processing
|
||||
- schema (dict): Data schema definition
|
||||
- table_name (str): Target table for predictions
|
||||
- model_id (str): ML model identifier
|
||||
- model_name (str): ML model name
|
||||
- input_filters (dict): Data quality filters
|
||||
- mlflow_transform_filters (dict): MLFlow transform filters
|
||||
- mlflow_predict_filters (dict): MLFlow prediction filters
|
||||
- model_retention (int): Model retention period in minutes
|
||||
- path_priority (list[str]): Decision path priority configuration
|
||||
- opc_output_config (dict): OPC server export configuration
|
||||
|
||||
Returns:
|
||||
None
|
||||
None: The workflow completes successfully when export workflow finishes
|
||||
|
||||
Raises:
|
||||
Exception: If any of the required parameters are missing or if the workflow fails.
|
||||
Exception: If any required parameters are missing or if the workflow fails
|
||||
during data processing, MLFlow operations, or workflow delegation
|
||||
|
||||
"""
|
||||
|
||||
metadata = input_data['metadata']
|
||||
@@ -47,6 +81,7 @@ class PredictionProcess():
|
||||
model_name = input_data['model_name']
|
||||
model_retention = input_data['model_retention']
|
||||
|
||||
# Get last timestamp for incremental processing
|
||||
last_timestamp = await workflow.execute_local_activity_method(
|
||||
Activities.get_last_timestamp,
|
||||
{
|
||||
@@ -57,6 +92,7 @@ class PredictionProcess():
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
)
|
||||
|
||||
# Apply input data quality gates
|
||||
gate_input = {
|
||||
**metadata,
|
||||
'filters': input_data['input_filters'],
|
||||
@@ -71,11 +107,13 @@ class PredictionProcess():
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
)
|
||||
|
||||
# Handle path decision based on filter results
|
||||
if await self.path_flag_handler(
|
||||
data, path_flag, input_data, confidence, last_timestamp, comment
|
||||
):
|
||||
return
|
||||
|
||||
# Request MLFlow model transformation
|
||||
response_data = await workflow.execute_local_activity_method(
|
||||
Activities.request_transform,
|
||||
{
|
||||
@@ -88,6 +126,7 @@ class PredictionProcess():
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
)
|
||||
|
||||
# Validate MLFlow transform response
|
||||
path_flag, confidence, comment = await workflow.execute_local_activity_method(
|
||||
Activities.mlflow_response_gate,
|
||||
{
|
||||
@@ -101,6 +140,7 @@ class PredictionProcess():
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
)
|
||||
|
||||
# Handle path decision based on transform validation
|
||||
if await self.path_flag_handler(
|
||||
data, path_flag, input_data, confidence, last_timestamp, comment
|
||||
):
|
||||
@@ -138,6 +178,7 @@ class PredictionProcess():
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
)
|
||||
|
||||
# Validate MLFlow prediction response
|
||||
path_flag, confidence, comment = await workflow.execute_local_activity_method(
|
||||
Activities.mlflow_response_gate,
|
||||
{
|
||||
@@ -151,11 +192,13 @@ class PredictionProcess():
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
)
|
||||
|
||||
# Handle path decision based on prediction validation
|
||||
if await self.path_flag_handler(
|
||||
data, path_flag, input_data, confidence, last_timestamp, comment
|
||||
):
|
||||
return
|
||||
|
||||
# Delegate to export workflow for data persistence
|
||||
await workflow.execute_child_workflow(
|
||||
'format_and_export_prediction',
|
||||
{
|
||||
@@ -174,30 +217,31 @@ class PredictionProcess():
|
||||
}
|
||||
)
|
||||
|
||||
async def path_flag_handler(self, data: dict[str, Any], path_flag: str,
|
||||
input_data: dict[str, Any], confidence: int,
|
||||
last_timestamp: str, comment: str):
|
||||
"""
|
||||
This function handles the path flag and the confidence of the prediction.
|
||||
It returns True if the prediction should be stopped. If path_flag is 'repeat',
|
||||
it repeats the last prediction.
|
||||
If path_flag is 'continue', it calls the write workflow. If path_flag is 'stop',
|
||||
it stops the prediction process.
|
||||
Args:
|
||||
data (dict[str, Any]): The data to be used for the prediction.
|
||||
path_flag (str): The path flag to determine the type of prediction to format
|
||||
confidence (int): The confidence of the prediction
|
||||
schema (str): The schema of the prediction
|
||||
table_name (str): The table name of the prediction
|
||||
model_id (int): The model id of the prediction
|
||||
last_timestamp (str): The timestamp of the last prediction
|
||||
model_name (str): The model name of the prediction
|
||||
model_retention (int): The model retention of the prediction
|
||||
comment (str): The comment of the prediction
|
||||
Returns:
|
||||
bool: True if the prediction should be stopped, False otherwise.
|
||||
async def path_flag_handler(self, data: dict, path_flag: str, input_data: dict,
|
||||
confidence: int, last_timestamp: str, comment: str) -> bool:
|
||||
"""
|
||||
Handle path decisions based on filter results and confidence levels.
|
||||
|
||||
This method determines the appropriate action based on the path flag
|
||||
returned by data quality filters. It can stop processing, continue,
|
||||
or repeat operations based on the configured path priority.
|
||||
|
||||
Args:
|
||||
data: Input data for processing
|
||||
path_flag: Path decision from filter (STOP, CONTINUE, REPEAT)
|
||||
input_data: Complete workflow input configuration
|
||||
confidence: Confidence level from filter validation
|
||||
last_timestamp: Last processed timestamp
|
||||
comment: Additional information about the filter result
|
||||
|
||||
Returns:
|
||||
bool: True if processing should stop, False to continue
|
||||
|
||||
Path Handling:
|
||||
- STOP: Terminates workflow execution
|
||||
- CONTINUE: Proceeds with normal processing
|
||||
- REPEAT: Repeats last prediction if available
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
|
||||
schema = input_data['schema']
|
||||
@@ -209,10 +253,10 @@ class PredictionProcess():
|
||||
path_flag = path_flag.upper() if path_flag else ''
|
||||
|
||||
if path_flag == 'STOP':
|
||||
# Stop processing and exit workflow
|
||||
return True
|
||||
|
||||
elif path_flag == 'REPEAT':
|
||||
# repeat last prediction
|
||||
# Repeat last prediction if available
|
||||
await workflow.execute_activity_method(
|
||||
Activities.repeat_last_prediction,
|
||||
{
|
||||
@@ -226,7 +270,6 @@ class PredictionProcess():
|
||||
start_to_close_timeout=timedelta(minutes=1),
|
||||
)
|
||||
return True
|
||||
|
||||
elif path_flag == 'CONTINUE':
|
||||
# call write workflow
|
||||
await workflow.execute_child_workflow(
|
||||
|
||||
@@ -3,6 +3,6 @@ psycopg2-binary
|
||||
sqlalchemy
|
||||
asyncua
|
||||
redis
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.4
|
||||
git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.5
|
||||
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
|
||||
prometheus-client
|
||||
|
||||
@@ -322,15 +322,68 @@ async def test_mlflow_content_gate_with_filter(gates_activity):
|
||||
gates_activity.send_notification.assert_called()
|
||||
|
||||
|
||||
def test_get_prediction_store_policy_invalid_policy(gates_activity):
|
||||
# Arrange
|
||||
prediction_store_policy = 'INVALID_POLICY'
|
||||
|
||||
# Act
|
||||
policy_type, policy_value = gates_activity.get_prediction_store_policy(
|
||||
prediction_store_policy, metadata)
|
||||
|
||||
# Assert
|
||||
assert policy_type == 'lts'
|
||||
assert policy_value == 1
|
||||
|
||||
|
||||
def test_get_prediction_store_policy_invalid_policy_value(gates_activity):
|
||||
# Arrange
|
||||
prediction_store_policy = 'abc:INVALID_VALUE'
|
||||
|
||||
# Act
|
||||
policy_type, policy_value = gates_activity.get_prediction_store_policy(
|
||||
prediction_store_policy, metadata)
|
||||
|
||||
# Assert
|
||||
assert policy_type == 'lts'
|
||||
assert policy_value == 1
|
||||
|
||||
|
||||
def test_get_prediction_store_policy_valid_policy_type(gates_activity):
|
||||
# Arrange
|
||||
prediction_store_policy = 'abc:1'
|
||||
|
||||
# Act
|
||||
policy_type, policy_value = gates_activity.get_prediction_store_policy(
|
||||
prediction_store_policy, metadata)
|
||||
|
||||
# Assert
|
||||
assert policy_type == 'lts'
|
||||
assert policy_value == 1
|
||||
|
||||
|
||||
def test_get_prediction_store_policy_valid_policy(gates_activity):
|
||||
# Arrange
|
||||
prediction_store_policy = 'erl:1'
|
||||
|
||||
# Act
|
||||
policy_type, policy_value = gates_activity.get_prediction_store_policy(
|
||||
prediction_store_policy, metadata)
|
||||
|
||||
# Assert
|
||||
assert policy_type == 'erl'
|
||||
assert policy_value == 1
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_format_prediction(gates_activity):
|
||||
async def test_format_prediction_no_timestamp(gates_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
**metadata,
|
||||
'data': {'prediction': [1], 'response_time': [0.1]},
|
||||
'timestamp': '2023-05-26 11:12:27',
|
||||
'model_id': 'test_model',
|
||||
'prediction_confidence': 0.9
|
||||
'prediction_confidence': 0.9,
|
||||
'prediction_store_policy': 'lts:1'
|
||||
}
|
||||
|
||||
# Act
|
||||
@@ -346,6 +399,83 @@ async def test_format_prediction(gates_activity):
|
||||
assert result['comments'] == {0: ""}
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_format_prediction_with_timestamp_erl(gates_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
**metadata,
|
||||
'data': {'prediction': [1, 2, 3],
|
||||
'response_time': [0.1, 0.2, 0.3],
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
|
||||
'model_id': 'test_model',
|
||||
'prediction_confidence': 0.9,
|
||||
'prediction_store_policy': 'erl:2'
|
||||
}
|
||||
|
||||
# Act
|
||||
result = await gates_activity.format_prediction(input_data)
|
||||
|
||||
# Assert
|
||||
assert result['prediction'] == {0: 2, 1: 1}
|
||||
assert result['response_time'] == {0: 0.2, 1: 0.1}
|
||||
assert result['timestamp'] == {
|
||||
0: '2023-05-26 11:12:28', 1: '2023-05-26 11:12:27'}
|
||||
assert result['model_id'] == {0: 'test_model', 1: 'test_model'}
|
||||
assert result['prediction_confidence'] == {0: 0.9, 1: 0.9}
|
||||
assert result['prediction_status'] == {0: 'Good', 1: 'Good'}
|
||||
assert result['comments'] == {0: "", 1: ""}
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_format_prediction_with_timestamp_lts(gates_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
**metadata,
|
||||
'data': {'prediction': [1, 2, 3],
|
||||
'response_time': [0.1, 0.2, 0.3],
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
|
||||
'model_id': 'test_model',
|
||||
'prediction_confidence': 0.9,
|
||||
'prediction_store_policy': 'lts:2'
|
||||
}
|
||||
|
||||
# Act
|
||||
result = await gates_activity.format_prediction(input_data)
|
||||
|
||||
# Assert
|
||||
assert result['prediction'] == {0: 3, 1: 2}
|
||||
assert result['response_time'] == {0: 0.3, 1: 0.2}
|
||||
assert result['timestamp'] == {
|
||||
0: '2023-05-26 11:12:29', 1: '2023-05-26 11:12:28'}
|
||||
assert result['model_id'] == {0: 'test_model', 1: 'test_model'}
|
||||
assert result['prediction_confidence'] == {0: 0.9, 1: 0.9}
|
||||
assert result['prediction_status'] == {0: 'Good', 1: 'Good'}
|
||||
assert result['comments'] == {0: "", 1: ""}
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_format_prediction_with_timestamp_invalid_policy(gates_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
**metadata,
|
||||
'data': {'prediction': [1, 2, 3],
|
||||
'response_time': [0.1, 0.2, 0.3],
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
|
||||
'model_id': 'test_model',
|
||||
'prediction_confidence': 0.9,
|
||||
'prediction_store_policy': 'lts:2'
|
||||
}
|
||||
gates_activity.get_prediction_store_policy = MagicMock(
|
||||
return_value=('invalid', 1))
|
||||
|
||||
try:
|
||||
result = await gates_activity.format_prediction(input_data)
|
||||
except ValueError as e:
|
||||
assert str(e) == "Invalid policy type: invalid"
|
||||
else:
|
||||
assert False, "Expected ValueError"
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
async def test_format_default_prediction(gates_activity):
|
||||
# Arrange
|
||||
|
||||
@@ -82,7 +82,8 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow):
|
||||
}
|
||||
|
||||
# Mock the transform response
|
||||
expected_response = {'prediction': [0.5, 0.6]}
|
||||
expected_response = {'prediction': [0.5, 0.6], 'timestamp': [
|
||||
'2024-01-01', '2024-01-02']}
|
||||
mlflow.model_monitoring_repository.transform.return_value = expected_response
|
||||
|
||||
mock_dataframe.return_value.sort_values.return_value = mock_dataframe.return_value
|
||||
@@ -98,7 +99,7 @@ async def test_request_transform(mock_max, mock_dataframe, mlflow):
|
||||
)
|
||||
mock_dataframe = mock_dataframe.return_value.pivot.return_value
|
||||
mock_dataframe.fillna.assert_called_once_with(np.nan, inplace=True)
|
||||
mock_dataframe.reset_index.assert_called_once()
|
||||
# mock_dataframe.reset_index.assert_called_once()
|
||||
mock_dataframe.columns.name = None
|
||||
|
||||
# Verify the response
|
||||
|
||||
@@ -35,7 +35,8 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction):
|
||||
"schema": "test_schema",
|
||||
"table_name": "test_table",
|
||||
"opc_servers": ["test_server"],
|
||||
"opc_output_config": {"test": "config"}
|
||||
"opc_output_config": {"test": "config"},
|
||||
"prediction_store_policy": "erl:1"
|
||||
}
|
||||
|
||||
await format_and_export_prediction.run(input_data)
|
||||
@@ -48,6 +49,7 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction):
|
||||
'timestamp': input_data['timestamp'],
|
||||
'model_id': input_data['model_id'],
|
||||
'prediction_confidence': input_data['prediction_confidence'],
|
||||
'prediction_store_policy': input_data['prediction_store_policy'],
|
||||
**metadata
|
||||
},
|
||||
retry_policy=ANY,
|
||||
|
||||
Reference in New Issue
Block a user