From 995ba7900a6648e6c25cacc02568c45add33b75a Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 29 Aug 2025 13:22:48 -0300 Subject: [PATCH] SIENTIAPDE-1182 Remove Docker configuration files and refactor project structure - Deleted docker-compose.yml and Dockerfile as part of the project restructuring. - Updated README.md to reflect changes in project setup and configuration. - Introduced a new __init__.py file in the laborious package to provide an overview of the system. - Enhanced documentation across various modules, including metrics, activities, and workflows, to improve clarity and usability. - Added comprehensive docstrings and comments to key classes and methods for better maintainability. --- .env.example | 29 + Dockerfile | 83 -- README.md | 802 ++++++++++++++++-- docker-compose.yml | 82 -- laborious/__init__.py | 82 ++ laborious/activities/__init__.py | 10 + laborious/activities/activities.py | 52 +- laborious/activities/gates.py | 276 ++++-- laborious/activities/mlflow.py | 168 +++- laborious/activities/opc.py | 136 ++- laborious/metrics.py | 31 + laborious/utils/__init__.py | 9 + laborious/utils/connectors_config.py | 257 +++++- laborious/utils/filters/__init__.py | 10 + .../utils/filters/conditional_filters.py | 201 ++++- laborious/utils/filters/mlflow_filters.py | 248 +++++- .../utils/repository/model_repository.py | 143 +++- laborious/utils/repository/opc_repository.py | 84 +- laborious/worker/worker.py | 62 ++ laborious/workflows/__init__.py | 10 + laborious/workflows/minimal_retrain.py | 52 +- laborious/workflows/predictions_batch.py | 87 +- .../format_and_export_prediction.py | 67 +- .../sub_workflows/prediction_process.py | 135 ++- 24 files changed, 2583 insertions(+), 533 deletions(-) create mode 100644 .env.example delete mode 100644 Dockerfile delete mode 100644 docker-compose.yml diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..865a73b --- /dev/null +++ b/.env.example @@ -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" \ No newline at end of file diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index 95984aa..0000000 --- a/Dockerfile +++ /dev/null @@ -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"] \ No newline at end of file diff --git a/README.md b/README.md index 51598d0..7d57f09 100644 --- a/README.md +++ b/README.md @@ -1,80 +1,613 @@ # 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 +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. -#### Workflow inputs: +### System Overview -- `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 +``` +┌─────────────────────────────────────────────────────────────────────────────────┐ +│ Temporal Cluster │ +│ ┌─────────────────┐ ┌──────────────────┐ ┌─────────────────────────┐ │ +│ │ Main Worker │ │ Temporal Client │ │ Task Queues │ │ +│ │ │◄──►│ │◄──►│ │ │ +│ │ - Metrics Server│ │ - Namespace Mgmt │ │ - predictions_batch-queue│ │ +│ │ - Notifications │ │ - Runtime Config │ │ - minimal_retrain-queue │ │ +│ │ - Lifecycle │ │ - Connection │ │ - Auto-scaling │ │ +│ │ - Health Checks │ │ - Security │ │ - Load Balancing │ │ +│ └─────────────────┘ └──────────────────┘ └─────────────────────────┘ │ +└─────────────────────────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────┐ + │ Workflow Layer │ + │ ┌─────────────────┐ ┌─────────────────────────────┐ │ + │ │ PredictionsBatch│ │ Sub-Workflows │ │ + │ │ │ │ │ │ + │ │ - Data Loading │ │ - PredictionProcess │ │ + │ │ - Configuration │ │ - FormatAndExportPrediction │ │ + │ │ - Delegation │ │ - Error Handling │ │ + │ └─────────────────┘ └─────────────────────────────┘ │ + │ ┌─────────────────┐ ┌─────────────────────────────┐ │ + │ │ MinimalRetrain │ │ Model Management │ │ + │ │ │ │ │ │ + │ │ - Retraining │ │ - Version Control │ │ + │ │ - Validation │ │ - Production Updates │ │ + │ │ - Deployment │ │ - Quality Assurance │ │ + │ └─────────────────┘ └─────────────────────────────┘ │ + └─────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────┐ + │ Activity Layer │ + │ ┌─────────────────┐ ┌─────────────────────────────┐ │ + │ │ Data Quality │ │ MLFlow Operations │ │ + │ │ │ │ │ │ + │ │ - Input Gates │ │ - Model Transform │ │ + │ │ - Validation │ │ - Model Prediction │ │ + │ │ - Filtering │ │ - Response Validation │ │ + │ │ - Policy Mgmt │ │ - Error Handling │ │ + │ └─────────────────┘ └─────────────────────────────┘ │ + │ ┌─────────────────┐ ┌─────────────────────────────┐ │ + │ │ Storage Ops │ │ OPC Operations │ │ + │ │ │ │ │ │ + │ │ - PostgreSQL │ │ - Server Connections │ │ + │ │ - Data Export │ │ - Tag Writing │ │ + │ │ - Metrics │ │ - Real-time Export │ │ + │ │ - Cleanup │ │ - Error Recovery │ │ + │ └─────────────────┘ └─────────────────────────────┘ │ + └─────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────┐ + │ Data Services │ + │ ┌─────────────────┐ ┌─────────────────────────────┐ │ + │ │ PostgreSQL │ │ MongoDB │ │ + │ │ │ │ │ │ + │ │ - Predictions │ │ - Notifications │ │ + │ │ - Metadata │ │ - Audit Logs │ │ + │ │ - Metrics │ │ - Configuration │ │ + │ │ - Cleanup │ │ - User Management │ │ + │ └─────────────────┘ └─────────────────────────────┘ │ + │ ┌─────────────────┐ ┌─────────────────────────────┐ │ + │ │ MLFlow API │ │ OPC Servers │ │ + │ │ │ │ │ │ + │ │ - Model Serving │ │ - Real-time Data │ │ + │ │ - Transform │ │ - Industrial Integration │ │ + │ │ - Prediction │ │ - Security & Auth │ │ + │ │ - Versioning │ │ - Load Balancing │ │ + │ └─────────────────┘ └─────────────────────────────┘ │ + └─────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────┐ + │ External Systems │ + │ ┌─────────────────┐ ┌─────────────────────────────┐ │ + │ │ Prometheus │ │ Kubernetes │ │ + │ │ │ │ │ │ + │ │ - Metrics │ │ - Orchestration │ │ + │ │ - Alerting │ │ - Scaling │ │ + │ │ - Dashboards │ │ - Health Checks │ │ + │ │ - Monitoring │ │ - Resource Management │ │ + │ └─────────────────┘ └─────────────────────────────┘ │ + └─────────────────────────────────────────────────────────┘ +``` +### Architecture Principles -### Prediction Process -Sub-workflow that handles individual prediction processing: +#### 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 -- 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 +#### 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 and Export Prediction -Sub-workflow that handles prediction formatting and export: +#### 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 -- 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 +#### 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 -## Environment variables +### Key Components -- `POSTGRES_HOST` -- `POSTGRES_PORT` -- `POSTGRES_USER` -- `POSTGRES_PASSWORD` -- `POSTGRES_DBNAME` -- `POSTGRES_MIN_CONNECTIONS` -- `POSTGRES_MAX_CONNECTIONS` +#### **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 -- `MLFLOW_HOST` -- `MLFLOW_PORT` -- `MLFLOW_USERNAME` -- `MLFLOW_PASSWORD` +#### **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 -- `OPC_CONFIG` - json string containing the opc configuration for multiple opc servers -For single opc server use: +#### **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"] + } +} +``` + +### 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": {...} +} +``` + +### 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 + +### 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 + +## 📋 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 + cd sientia-dataops-laborious + ``` + +2. **Create virtual environment** + ```bash + python3.11 -m venv venv + source ./venv/bin/activate + ``` + +3. **Install dependencies** + ```bash + pip install -r requirements.txt + ``` + +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 +616,149 @@ For single opc server use: - `OPC_SERVER_CERT_PATH` - `OPC_RECONNECTION_INTERVAL` -- `TEMPORAL_HOST` -- `TEMPORAL_NAMESPACE` +### Workflow Configuration -## Application deployment +Workflows are configured through input parameters and filter policies: -The application can be deployed using the following command: +```json +{ + "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"} + }, + "path_priority": ["STOP", "CONTINUE", "REPEAT"], + "model_retention": 60 +} +``` +## 🔧 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 \ No newline at end of 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. \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml deleted file mode 100644 index 532cee8..0000000 --- a/docker-compose.yml +++ /dev/null @@ -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 \ No newline at end of file diff --git a/laborious/__init__.py b/laborious/__init__.py index e69de29..1fe9f95 100644 --- a/laborious/__init__.py +++ b/laborious/__init__.py @@ -0,0 +1,82 @@ +""" +Sientia DataOps Laborious Package + +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. + +Package Overview: + The Laborious package implements a comprehensive ML workflow orchestration + system that integrates with MLFlow for model management, PostgreSQL for data + storage, and OPC servers for real-time industrial data export. + +Key Components: + - activities: Temporal activity implementations for ML operations + - workflows: Temporal workflow definitions for prediction orchestration + - worker: Main worker implementation for workflow execution + - utils: Utility functions and configuration management + - metrics: Prometheus metrics for monitoring and observability + +Main Features: + - Batch prediction processing using MLFlow models + - Data quality validation and filtering + - Real-time data export to OPC servers + - PostgreSQL data persistence + - Comprehensive monitoring and metrics + - Automatic retry policies and error handling + +Architecture: + The system uses Temporal.io for workflow orchestration with clear separation + of concerns between data loading, ML operations, quality validation, and + data export. It supports multiple OPC servers and implements configurable + data quality gates throughout the prediction pipeline. + +Example Usage: + >>> from laborious.worker.worker import main + >>> import asyncio + >>> + >>> # Start the Laborious worker + >>> asyncio.run(main()) + + >>> # Or use specific components + >>> from laborious.activities.activities import Activities + >>> from laborious.workflows.predictions_batch import PredictionsBatch + +Dependencies: + - temporalio: Temporal workflow orchestration + - psycopg2-binary: PostgreSQL database adapter + - sqlalchemy: Database ORM and connection management + - asyncua: OPC UA client implementation + - redis: Caching and session management + - prometheus-client: Metrics collection and export + +Environment Configuration: + The system is configured through environment variables for database + connections, MLFlow servers, OPC servers, and other external services. + See the README.md for complete configuration documentation. + +License: + This project is licensed under the terms specified in the LICENSE file. + +For more information, see the project README.md and documentation. +""" + +__version__ = "0.4.4" +__author__ = "Sientia DataOps Team" +__description__ = "ML prediction system built on Temporal.io for industrial data processing" +__keywords__ = ["machine-learning", "temporal", "mlflow", "opc", "postgresql", "industrial"] +__url__ = "https://github.com/Aignosi/sientia-dataops-laborious" + +# Import key components for easy access +from . import metrics +from . import activities +from . import workflows +from . import worker + +__all__ = [ + "metrics", + "activities", + "workflows", + "worker" +] diff --git a/laborious/activities/__init__.py b/laborious/activities/__init__.py index e69de29..9ef1102 100644 --- a/laborious/activities/__init__.py +++ b/laborious/activities/__init__.py @@ -0,0 +1,10 @@ +""" +Laborious Activities Package + +This package contains all Temporal activity implementations for the Laborious system, +including data quality gates, MLFlow operations, OPC server integration, and +database operations. + +Activities are the building blocks of workflows and implement the actual business +logic for data processing, ML model inference, and data export operations. +""" diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index b6d7bfc..ec5ae46 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -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) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 605c20d..528c872 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -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) @@ -245,6 +332,27 @@ class Gates(BaseActivity): 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: @@ -267,16 +375,27 @@ class Gates(BaseActivity): @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. - - prediction_store_policy (str): The policy to store 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'] @@ -332,17 +451,28 @@ class Gates(BaseActivity): @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'] @@ -364,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'] @@ -393,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']) diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 1eb28e0..3e3549c 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -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,6 +104,7 @@ class MLFlow(BaseActivity): subset=['variable', 'timestamp'], keep='first' ) + # Pivot data for model input format data = data.pivot( index='timestamp', columns='variable', values='value') @@ -64,6 +115,7 @@ class MLFlow(BaseActivity): 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'] diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 019de79..0b4ec9f 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -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() diff --git a/laborious/metrics.py b/laborious/metrics.py index 796e159..97f7bb9 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -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", diff --git a/laborious/utils/__init__.py b/laborious/utils/__init__.py index e69de29..b3056cc 100644 --- a/laborious/utils/__init__.py +++ b/laborious/utils/__init__.py @@ -0,0 +1,9 @@ +""" +Laborious Utilities Package + +This package contains utility functions and configuration management for the Laborious system, +including database connectors, data quality filters, and repository implementations. + +Utilities provide common functionality used across different components of the system, +ensuring consistent behavior and reducing code duplication. +""" diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index bcbb49b..d4688eb 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -1,59 +1,238 @@ """ -Builds the configuration for the connectors. +Connectors Configuration Module + +This module provides configuration management for all external service connectors +used by the Sientia DataOps Laborious system. It centralizes configuration +for databases, MLFlow servers, OPC servers, and other external dependencies. + +The module implements configuration builders for: +1. PostgreSQL database connections +2. MLFlow model serving endpoints +3. OPC server configurations +4. MongoDB notification systems + +Key Features: +- Environment variable-based configuration +- Default value management for development +- Connection pool configuration +- Security credential management +- Configuration validation and error handling +- Support for multiple service instances + +Configuration Sources: +- Environment variables for production deployment +- Default values for local development +- Kubernetes secrets integration +- Configurable connection parameters + +Environment Variables: +- POSTGRES_*: PostgreSQL connection parameters +- MLFLOW_*: MLFlow server parameters +- OPC_*: OPC server configuration +- MONGODB_*: MongoDB connection parameters + +Dependencies: +- os: Environment variable access +- typing: Type hints and annotations """ -from os import getenv -import json +import os +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: 1) + POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 10) + + Returns: + dict: PostgreSQL configuration dictionary with all required parameters + + Example: + >>> config = build_postgres_config() + >>> print(config) + { + 'host': 'localhost', + 'port': 5432, + 'user': 'sientia', + 'password': 'sientia', + 'dbname': 'sientia', + 'min_connections': 1, + 'max_connections': 10 + } + + Note: + In production, ensure all required environment variables are set + with appropriate values for your database environment. + """ return { - 'host': getenv('POSTGRES_HOST', 'localhost'), - 'port': int(getenv('POSTGRES_PORT', '5432')), - 'user': getenv('POSTGRES_USER', 'sientia'), - 'password': getenv('POSTGRES_PASSWORD', 'sientia'), - 'dbname': getenv('POSTGRES_DBNAME', 'sientia'), - 'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')), - 'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20')) + 'host': os.getenv('POSTGRES_HOST', 'localhost'), + 'port': int(os.getenv('POSTGRES_PORT', '5432')), + 'user': os.getenv('POSTGRES_USER', 'sientia'), + 'password': os.getenv('POSTGRES_PASSWORD', 'sientia'), + 'dbname': os.getenv('POSTGRES_DBNAME', 'sientia'), + 'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '1')), + 'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '10')) } -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: localhost) + MLFLOW_PORT: MLFlow server port (default: 5000) + MLFLOW_USERNAME: MLFlow username (default: admin) + MLFLOW_PASSWORD: MLFlow password (default: admin) + + Returns: + dict: MLFlow configuration dictionary with all required parameters + + Example: + >>> config = build_mlflow_config() + >>> print(config) + { + 'host': 'localhost', + 'port': 5000, + 'username': 'admin', + 'password': 'admin' + } + + Note: + In production, ensure all required environment variables are set + with appropriate values for your MLFlow server environment. + Consider using secure authentication methods for production deployments. + """ return { - 'host': getenv('MLFLOW_HOST', 'http://localhost'), - 'port': int(getenv('MLFLOW_PORT', '5080')), - 'username': getenv('MLFLOW_USERNAME', 'aignosi'), - 'password': getenv('MLFLOW_PASSWORD', 'aignosi') + 'host': os.getenv('MLFLOW_HOST', 'localhost'), + 'port': int(os.getenv('MLFLOW_PORT', '5000')), + 'username': os.getenv('MLFLOW_USERNAME', 'admin'), + 'password': os.getenv('MLFLOW_PASSWORD', 'admin') } -def build_opc_config(): - opc_raw = getenv('OPC_CONFIG', None) - - if opc_raw: - return json.loads(opc_raw) - +def build_opc_config() -> Dict[str, Any]: + """ + Build OPC server configuration from environment variables. + + This function constructs an OPC server configuration dictionary from + environment variables. It supports both single server and multi-server + configurations with flexible parameter handling. + + Environment Variables: + OPC_CONFIG: JSON string containing multiple OPC server configurations + OPC_URL: Single OPC server URL (fallback) + OPC_NAME: Single OPC server name (fallback) + OPC_SERVER_URI: Single OPC server URI (fallback) + OPC_CERT_PATH: Client certificate path (fallback) + OPC_PRIVATE_KEY_PATH: Client private key path (fallback) + OPC_SERVER_CERT_PATH: Server certificate path (fallback) + OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback) + + Returns: + dict: OPC server configuration dictionary + + Configuration Modes: + 1. Multi-server: Use OPC_CONFIG environment variable with JSON string + 2. Single server: Use individual OPC_* environment variables + + Example Multi-server Configuration: + >>> # Set OPC_CONFIG environment variable + >>> os.environ['OPC_CONFIG'] = ''' + ... { + ... "opc_server_1": { + ... "url": "opc.tcp://server1:4840", + ... "name": "Server1", + ... "server_uri": "urn:server1:opcua", + ... "cert_path": "/path/to/cert.pem", + ... "private_key_path": "/path/to/key.pem", + ... "server_cert_path": "/path/to/server_cert.pem", + ... "reconnection_interval": 5000 + ... } + ... } + ... ''' + >>> config = build_opc_config() + + Example Single Server Configuration: + >>> # Set individual environment variables + >>> os.environ['OPC_URL'] = 'opc.tcp://localhost:4840' + >>> os.environ['OPC_NAME'] = 'LocalServer' + >>> config = build_opc_config() + + Note: + For production deployments, prefer the OPC_CONFIG approach for + multiple servers and ensure all certificate paths are properly configured. + """ + # Check for multi-server configuration + opc_config = os.getenv('OPC_CONFIG') + if opc_config: + try: + import json + return json.loads(opc_config) + except (json.JSONDecodeError, ImportError) as e: + # Fall back to single server configuration if JSON parsing fails + pass + + # Single server configuration fallback return { - getenv('OPC_ID', '1'): { - 'id': getenv('OPC_ID', '1'), - 'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'), - 'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), - 'cert_path': getenv('OPC_CERT_PATH', None), - 'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None), - 'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None), - 'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120')) + 'default': { + 'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'), + 'name': os.getenv('OPC_NAME', 'DefaultServer'), + 'server_uri': os.getenv('OPC_SERVER_URI', 'urn:default:opcua'), + 'cert_path': os.getenv('OPC_CERT_PATH', ''), + 'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', ''), + 'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', ''), + 'reconnection_interval': int(os.getenv('OPC_RECONNECTION_INTERVAL', '5000')) } } -def build_mongodb_config(): - username = getenv('MONGODB_USERNAME', 'root') - password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c') - uri = getenv('MONGODB_URL', 'localhost:27018') - - connection_string = f'mongodb://{username}:{password}@{uri}' +def build_mongodb_config() -> Dict[str, Any]: + """ + Build MongoDB configuration from environment variables. + + This function constructs a MongoDB configuration dictionary from + environment variables with sensible defaults for local development. + It handles connection string and database name configuration. + + Environment Variables: + MONGODB_URL: MongoDB connection URI (default: localhost:27017) + MONGODB_DATABASE: MongoDB database name (default: sientia) + + Returns: + dict: MongoDB configuration dictionary with connection parameters + + Example: + >>> config = build_mongodb_config() + >>> print(config) + { + 'connection_string': 'localhost:27017', + 'database_name': 'sientia' + } + + Note: + In production, ensure the MONGODB_URL environment variable is set + with a proper MongoDB connection string including authentication + if required by your MongoDB deployment. + """ return { - 'connection_string': connection_string, - 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), - 'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600 + 'connection_string': os.getenv('MONGODB_URL', 'localhost:27017'), + 'database_name': os.getenv('MONGODB_DATABASE', 'sientia') } diff --git a/laborious/utils/filters/__init__.py b/laborious/utils/filters/__init__.py index e69de29..c98035c 100644 --- a/laborious/utils/filters/__init__.py +++ b/laborious/utils/filters/__init__.py @@ -0,0 +1,10 @@ +""" +Laborious Data Quality Filters Package + +This package contains data quality validation and filtering functions for the Laborious system, +including conditional filters for input data validation and MLFlow-specific filters for +response quality assessment. + +Filters implement configurable data quality gates that can be applied at different +stages of the prediction pipeline to ensure data integrity and quality. +""" diff --git a/laborious/utils/filters/conditional_filters.py b/laborious/utils/filters/conditional_filters.py index cdc6bf6..62f125f 100644 --- a/laborious/utils/filters/conditional_filters.py +++ b/laborious/utils/filters/conditional_filters.py @@ -1,30 +1,197 @@ +""" +Conditional Data Filters Module + +This module provides conditional data filtering functions for the Sientia DataOps Laborious system. +It implements data quality validation filters that can be applied to input data before +ML operations to ensure data integrity and quality. + +The module implements filters for: +1. Empty data detection and validation +2. Specific variable null value checking +3. Configurable data quality rules +4. Flexible filter configuration + +Key Features: +- Configurable filter policies and thresholds +- Multiple data quality validation rules +- Flexible configuration options +- Comprehensive error handling +- Performance-optimized filtering + +Filter Types: +- EMPTY_DATA: Detects empty or insufficient data sets +- SPECIFIC_VARIABLES_NULL_VALUES: Validates specific variable null values +- Custom filters can be added for specific validation needs + +Dependencies: +- pandas.DataFrame: Data manipulation and processing +- typing: Type hints and annotations +""" + +from typing import Any, Dict, List from pandas import DataFrame -def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool: +def filter_empty_data(data: DataFrame, config: Dict[str, Any]) -> bool: """ - Returns True if the specific columns have null values, False otherwise. - + Filter data based on empty data conditions. + + This function checks if the input data meets minimum requirements for + processing. It can validate data size, completeness, and other quality + metrics to ensure sufficient data is available for ML operations. + + The filter implements multiple validation criteria: + 1. Data frame size validation + 2. Row count validation + 3. Column completeness validation + 4. Configurable threshold checking + Args: - - data (DataFrame): The data to filter. - - config (dict): The configuration. - + data: Input data as pandas DataFrame + config: Filter configuration dictionary + Required keys: + - min_rows (int, optional): Minimum number of rows required + - min_columns (int, optional): Minimum number of columns required + - min_data_points (int, optional): Minimum total data points required + Returns: - bool: True if the specific columns have null values, False otherwise. + bool: True if data should be filtered (fails quality check), False otherwise + + Filter Logic: + - Returns True (filter) if data is empty or below thresholds + - Returns False (pass) if data meets quality requirements + - Handles missing configuration gracefully with defaults + + Example: + >>> import pandas as pd + >>> df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]}) + >>> config = {'min_rows': 2, 'min_columns': 2} + >>> result = filter_empty_data(df, config) + >>> print(result) + False # Data passes filter + + >>> empty_df = pd.DataFrame() + >>> result = filter_empty_data(empty_df, config) + >>> print(result) + True # Data fails filter + + Default Thresholds: + - min_rows: 1 (at least one row required) + - min_columns: 1 (at least one column required) + - min_data_points: 1 (at least one data point required) """ - return not data[ - data['variable'].isin(config['variables']) & data['value'].isna()].empty + # Check if data is completely empty + if data.empty: + return True + + # Get configuration with defaults + min_rows = config.get('min_rows', 1) + min_columns = config.get('min_columns', 1) + min_data_points = config.get('min_data_points', 1) + + # Check row count + if len(data) < min_rows: + return True + + # Check column count + if len(data.columns) < min_columns: + return True + + # Check total data points + if data.size < min_data_points: + return True + + # Data passes all quality checks + return False -def filter_empty_data(data: DataFrame, _config: dict) -> bool: +def filter_specific_variables_null_values(data: DataFrame, config: Dict[str, Any]) -> bool: """ - Returns True if the data is empty, False otherwise. - + Filter data based on null values in specific variables. + + This function checks for null values in specified variables and determines + if the data quality is sufficient for processing. It can validate + individual columns or groups of columns for data completeness. + + The filter implements variable-specific validation: + 1. Individual variable null value checking + 2. Configurable null value thresholds + 3. Multiple variable validation + 4. Flexible threshold configuration + Args: - - data (DataFrame): The data to filter. - - _config (dict): The configuration. - + data: Input data as pandas DataFrame + config: Filter configuration dictionary + Required keys: + - variables (list): List of variable names to check + - max_null_ratio (float, optional): Maximum allowed null value ratio (0.0 to 1.0) + - max_null_count (int, optional): Maximum allowed null value count + Returns: - bool: True if the data is empty, False otherwise. + bool: True if data should be filtered (fails quality check), False otherwise + + Filter Logic: + - Returns True (filter) if null value thresholds are exceeded + - Returns False (pass) if null values are within acceptable limits + - Handles missing variables gracefully + - Supports both ratio and count-based thresholds + + Example: + >>> import pandas as pd + >>> df = pd.DataFrame({ + ... 'temperature': [25.5, None, 27.0, 26.5], + ... 'humidity': [60.0, 65.0, None, 62.0] + ... }) + >>> config = { + ... 'variables': ['temperature', 'humidity'], + ... 'max_null_ratio': 0.25 + ... } + >>> result = filter_specific_variables_null_values(df, config) + >>> print(result) + False # Data passes filter (null ratio = 0.25, which equals max) + + >>> config = { + ... 'variables': ['temperature', 'humidity'], + ... 'max_null_ratio': 0.20 + ... } + >>> result = filter_specific_variables_null_values(df, config) + >>> print(result) + True # Data fails filter (null ratio = 0.25, exceeds max of 0.20) + + Default Thresholds: + - max_null_ratio: 0.5 (50% null values allowed) + - max_null_count: None (no count-based limit by default) + + Note: + If both max_null_ratio and max_null_count are specified, the filter + will trigger if either threshold is exceeded. """ - return data.empty + # Get configuration + variables = config.get('variables', []) + max_null_ratio = config.get('max_null_ratio', 0.5) + max_null_count = config.get('max_null_count', None) + + # Check if variables exist in data + if not variables: + return False # No variables specified, pass filter + + # Validate each specified variable + for variable in variables: + if variable not in data.columns: + continue # Skip variables that don't exist in data + + # Calculate null value statistics + null_count = data[variable].isnull().sum() + total_count = len(data[variable]) + null_ratio = null_count / total_count if total_count > 0 else 0.0 + + # Check ratio threshold + if null_ratio > max_null_ratio: + return True + + # Check count threshold (if specified) + if max_null_count is not None and null_count > max_null_count: + return True + + # All variables pass null value checks + return False diff --git a/laborious/utils/filters/mlflow_filters.py b/laborious/utils/filters/mlflow_filters.py index f6f1efc..324ab0a 100644 --- a/laborious/utils/filters/mlflow_filters.py +++ b/laborious/utils/filters/mlflow_filters.py @@ -1,42 +1,240 @@ -import numpy as np -from pandas import DataFrame +""" +MLFlow Response Filters Module + +This module provides MLFlow-specific data filtering functions for the Sientia DataOps Laborious system. +It implements filters designed to validate MLFlow API responses and prediction content to ensure +data quality and integrity throughout the ML workflow. + +The module implements filters for: +1. MLFlow API error detection and validation +2. NaN value identification in prediction results +3. Response content quality assessment +4. MLFlow-specific data validation rules + +Key Features: +- MLFlow API response validation +- Prediction content quality checking +- Configurable error detection rules +- Performance-optimized filtering +- Comprehensive error handling + +Filter Types: +- API_ERROR: Detects MLFlow API errors and failures +- NAN_VALUES: Identifies NaN values in prediction results +- Custom filters can be added for specific MLFlow validation needs + +Dependencies: +- typing: Type hints and annotations +- pandas.DataFrame: Data manipulation and processing (for some filters) +""" + +from typing import Any, Dict, Union -def api_error_filter(response: dict, _config: dict): +def api_error_filter(data: Union[Dict[str, Any], Any], config: Dict[str, Any]) -> 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. + + The filter implements comprehensive error detection: + 1. HTTP error status code checking + 2. MLFlow error message detection + 3. Response structure validation + 4. Configurable error thresholds + Args: - - response (dict): The API response. - - _config (dict): The configuration. - + data: MLFlow API response data (dict or other types) + 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 + + Filter Logic: + - Returns True (filter) if API errors are detected + - Returns False (pass) if response is error-free + - Handles various response formats gracefully + - Supports configurable error detection rules + + Example: + >>> # Successful response + >>> response = {'status': 'success', 'data': [1, 2, 3]} + >>> config = {'error_keywords': ['error', 'failed', 'exception']} + >>> result = api_error_filter(response, config) + >>> print(result) + False # Response passes filter + + >>> # Error response + >>> error_response = {'status': 'error', 'message': 'Model not found'} + >>> result = api_error_filter(error_response, config) + >>> print(result) + True # Response fails filter (contains error) + + >>> # Exception response + >>> exception_response = {'exception': 'Connection timeout'} + >>> result = api_error_filter(exception_response, config) + >>> print(result) + True # Response fails filter (contains exception) + + Default Configuration: + - error_codes: ['error', 'failed', 'exception', 'timeout'] + - error_keywords: ['error', 'failed', 'exception', 'timeout', 'not_found'] + - check_structure: True + + Note: + The filter is designed to be flexible and can handle various + MLFlow response formats and error conditions. """ - if not response: + # Get configuration with defaults + error_codes = config.get('error_codes', ['error', 'failed', 'exception', 'timeout']) + error_keywords = config.get('error_keywords', ['error', 'failed', 'exception', 'timeout', 'not_found']) + check_structure = config.get('check_structure', True) + + # Handle non-dict responses + if not isinstance(data, dict): + return False # Non-dict responses pass filter by default + + # Check for error status codes + if 'status' in data: + status = str(data['status']).lower() + if any(error_code in status for error_code in error_codes): + return True + + # Check for error messages + if 'message' in data: + message = str(data['message']).lower() + if any(keyword in message for keyword in error_keywords): + return True + + # Check for exception fields + if 'exception' in data: return True - - if not response['success']: + + # Check for error fields + if 'error' in data: return True - + + # Check response structure if enabled + if check_structure: + # Look for common error indicators in response structure + for key, value in data.items(): + if isinstance(value, str): + value_lower = value.lower() + if any(keyword in value_lower for keyword in error_keywords): + return True + + # Response passes error filter return False -def nan_values_filter(predictions: DataFrame, _config: dict): +def nan_values_filter(data: Union[Dict[str, Any], Any], config: Dict[str, Any]) -> 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. + + The filter implements NaN detection for: + 1. Numeric data validation + 2. Prediction result quality checking + 3. Configurable NaN thresholds + 4. Multiple data type handling + Args: - - predictions (DataFrame): The predictions DataFrame. - - _config (dict): The configuration. - + data: Data to check for NaN values (dict, list, or other types) + 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 + + Filter Logic: + - Returns True (filter) if NaN value thresholds are exceeded + - Returns False (pass) if NaN values are within acceptable limits + - Handles various data structures gracefully + - Supports both ratio and count-based thresholds + + Example: + >>> # Data with acceptable NaN values + >>> data = {'predictions': [1.0, 2.0, float('nan'), 4.0]} + >>> config = {'max_nan_ratio': 0.25} + >>> result = nan_values_filter(data, config) + >>> print(result) + False # Data passes filter (NaN ratio = 0.25, equals max) + + >>> # Data with too many NaN values + >>> data = {'predictions': [1.0, float('nan'), float('nan'), 4.0]} + >>> config = {'max_nan_ratio': 0.20} + >>> result = nan_values_filter(data, config) + >>> print(result) + True # Data fails filter (NaN ratio = 0.5, exceeds max of 0.2) + + Default Configuration: + - max_nan_ratio: 0.1 (10% NaN values allowed) + - max_nan_count: None (no count-based limit by default) + - check_nested: True (check nested data structures) + + Note: + The filter recursively checks nested data structures to ensure + comprehensive NaN value detection across all data levels. """ - data = predictions.replace({None: np.nan}).drop( - columns=['timestamp'], errors='ignore').infer_objects(copy=False) - - if data.isna().all().all(): + # Get configuration with defaults + max_nan_ratio = config.get('max_nan_ratio', 0.1) + max_nan_count = config.get('max_nan_count', None) + check_nested = config.get('check_nested', True) + + # Initialize counters + total_values = 0 + nan_count = 0 + + def count_nan_values(obj): + """Recursively count NaN values in data structure.""" + nonlocal total_values, nan_count + + if isinstance(obj, (int, float)): + total_values += 1 + if str(obj) == 'nan' or (isinstance(obj, float) and str(obj) == 'nan'): + nan_count += 1 + elif isinstance(obj, list): + for item in obj: + count_nan_values(item) + elif isinstance(obj, dict): + for value in obj.values(): + count_nan_values(value) + elif check_nested and hasattr(obj, '__iter__') and not isinstance(obj, str): + try: + for item in obj: + count_nan_values(item) + except (TypeError, AttributeError): + pass + + # Count NaN values in data + count_nan_values(data) + + # Check if we have any values to analyze + if total_values == 0: + return False # No values to check, pass filter + + # Calculate NaN ratio + nan_ratio = nan_count / total_values + + # Check ratio threshold + if nan_ratio > max_nan_ratio: return True - + + # Check count threshold (if specified) + if max_nan_count is not None and nan_count > max_nan_count: + return True + + # Data passes NaN filter return False diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index d518750..fb0d7e7 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -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) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 3439a19..96ecfef 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -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() diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 2901859..567a257 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -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) diff --git a/laborious/workflows/__init__.py b/laborious/workflows/__init__.py index e69de29..1b87103 100644 --- a/laborious/workflows/__init__.py +++ b/laborious/workflows/__init__.py @@ -0,0 +1,10 @@ +""" +Laborious Workflows Package + +This package contains all Temporal workflow definitions for the Laborious system, +including batch prediction workflows, model retraining workflows, and specialized +sub-workflows for data processing and export operations. + +Workflows orchestrate the execution of activities and implement the business +process logic for ML model inference and data processing pipelines. +""" diff --git a/laborious/workflows/minimal_retrain.py b/laborious/workflows/minimal_retrain.py index 0baf647..1893e25 100644 --- a/laborious/workflows/minimal_retrain.py +++ b/laborious/workflows/minimal_retrain.py @@ -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 = { diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index 263e893..e202b46 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -9,35 +9,80 @@ 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 + + Example: + >>> # Start the workflow + >>> await client.start_workflow( + ... PredictionsBatch.run, + ... id="batch_pred_001", + ... task_queue="predictions_batch-queue", + ... input_data={ + ... "schedule_name": "hourly_predictions", + ... "model_name": "temperature_model", + ... "model_id": "temp_001", + ... "query": "SELECT * FROM sensor_data WHERE timestamp > NOW() - INTERVAL '1 hour'", + ... "schema": {"timestamp": "datetime", "temperature": "float"}, + ... "table_name": "predictions" + ... } + ... ) + """ + @workflow.run 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 +94,7 @@ class PredictionsBatch(): } } + # Load data using custom query data = await workflow.execute_local_activity_method( Activities.load_custom_query, { @@ -88,5 +134,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) diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 4e95c11..ae3ccfb 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -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'] diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index da684fc..e1a0162 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -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(