From 276561a8fce0efef0689eaf4413e5e6f0a7247bb Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 27 Aug 2026 14:39:04 -0300 Subject: [PATCH] Add README.md --- README.md | 1423 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1423 insertions(+) create mode 100644 README.md diff --git a/README.md b/README.md new file mode 100644 index 0000000..48c75d6 --- /dev/null +++ b/README.md @@ -0,0 +1,1423 @@ +# Sientia DataOps Laborious + +A comprehensive, Temporal-based ML orchestration system for industrial data processing and model inference. Laborious delivers enterprise-grade batch prediction, model management, optional real-time export (OPC and PI Web API), and automated retraining with strong data quality validation and observability. + + +## πŸ“‘ Table of Contents + +- [Features](#features) + - [Core Functionality](#core-functionality) + - [Advanced Capabilities](#advanced-capabilities) + - [Development & Quality Assurance](#development--quality-assurance) +- [Architecture](#architecture) + - [Architecture Principles](#architecture-principles) + - [Key Components](#key-components) + - [Data Flow Architecture](#data-flow-architecture) + - [Security Architecture](#security-architecture) +- [Workflows](#workflows) + - [Predictions Batch Workflow](#1-predictions-batch-workflow-predictions_batchpy) + - [Prediction Process Workflow](#2-prediction-process-workflow-prediction_processpy) + - [Format and Export Prediction Workflow](#3-format-and-export-prediction-workflow-format_and_export_predictionpy) + - [Minimal Retrain Workflow](#4-minimal-retrain-workflow-minimal_retrainpy) + - [Drift Workflow](#5-drift-workflow-driftpy) + - [Simple Metrics Workflow](#6-simple-metrics-workflow-simple_metricspy) + - [Import Model Workflow](#7-import-model-workflow-import_modelpy) +- [Installation & Setup](#installation--setup) + - [Prerequisites](#prerequisites) + - [Environment Setup](#environment-setup) + - [Temporal Namespace Setup](#temporal-namespace-setup) + - [Local Development Setup](#local-development-setup) +- [How to Run](#how-to-run) + - [Running the Laborious Application](#running-the-laborious-application) + - [Running Tests and Coverage](#running-tests-and-coverage) + - [Manual Test Execution](#manual-test-execution) + - [Manual Application Execution](#manual-application-execution) + - [Running One Model Import by Hand](#running-one-model-import-by-hand) +- [Code Quality & Validation](#code-quality--validation) + - [Overview](#overview) + - [Validation Tools](#validation-tools) + - [Tools Installation](#tools-installation) + - [Complete Validation](#complete-validation) + - [Automatic Fixes](#automatic-fixes) + - [Configuration](#configuration) + - [CI/CD Integration](#cicd-integration) + - [Best Practices](#best-practices) +- [Testing](#testing) + - [Test Structure](#test-structure) + - [Test Execution](#test-execution) + - [End-to-End Tests](#end-to-end-tests-e2e) +- [Monitoring and Metrics](#monitoring-and-metrics) + - [Application Health Metrics](#application-health-metrics) + - [Prediction Operation Metrics](#prediction-operation-metrics) + - [OPC Export Metrics](#opc-export-metrics) + - [Data Quality Metrics](#data-quality-metrics) +- [OPC UA Communication](#opc-ua-communication) +- [Configuration](#configuration-1) + - [Environment Variables](#environment-variables) + - [OPC Configuration](#opc-configuration) + - [Workflow Configuration](#workflow-configuration) +- [Development](#development) + - [Project Structure](#project-structure) + - [Adding New Features](#adding-new-features) +- [Troubleshooting](#troubleshooting) + - [Common Issues](#common-issues) + - [Debug Mode](#debug-mode) +- [Performance Tuning](#performance-tuning) + - [Key Parameters](#key-parameters) + - [Scaling Considerations](#scaling-considerations) +- [Contributing](#contributing) + - [Code Quality Standards](#code-quality-standards) +- [License](#license) +- [Support](#support) + +## Features + +### Core Functionality +- **Batch Prediction Processing**: High-throughput ML inference using MLFlow models +- **Temporal Workflow Orchestration**: Robust workflow management with retries and fault tolerance +- **Data Quality Gates**: Configurable filtering for input data and MLFlow API responses +- **Multi-Model Support**: Flexible model management with retention and versioning +- **Optional Real-time Export**: PostgreSQL persistence, OPC server integration, and PI Web API integration for industrial systems +- **Comprehensive Monitoring**: Prometheus metrics and structured logging for observability + +### Advanced Capabilities +- **Incremental Data Processing**: Timestamp-based loading to avoid reprocessing +- **Configurable Data Retention**: Model retention policies with automatic cleanup +- **MinIO Payload Offload**: Automatic offload of large DataFrames to MinIO with retention cleanup +- **Data Drift Detection**: Univariate and multivariate drift monitoring against reference data +- **Regression Metrics**: Automated RMSE, MSE, MAE, RΒ² calculation and export +- **Notification System**: Integrated alerting via MongoDB +- **Scalable Architecture**: Kubernetes-ready with horizontal scaling +- **Model Retraining**: Automated retraining workflows with production model updates + +### Development & Quality Assurance +- **Code Quality Tools**: Ruff (lint/format), mypy (types), Bandit (security) +- **Automated Validation**: CI quality gates and individual tool commands +- **Comprehensive Testing**: pytest with async support and high coverage +- **Type Safety**: Static type checking with mypy +- **Coverage Visualization**: Coverage Gutters integration + +## Architecture + +Laborious uses a Temporal-based architecture with strong separation of concerns and defensive error handling for production ML. + +### Architecture Principles + +#### 1. **Separation of Concerns** +- **Worker Layer**: Temporal workers, task queues, lifecycle +- **Workflow Layer**: Business orchestration and coordination +- **Activity Layer**: External system interactions and isolated operations +- **Data Layer**: Persistence, caching, connectors + +#### 2. **Fault Tolerance & Resilience** +- **Automatic Retry Policies** for transient failures +- **Graceful Degradation** and circuit breaking for dependencies +- **Detailed Error Handling** with notifications + +#### 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 + +#### 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 + +### Key Components + +#### **Worker (`laborious/worker/worker.py`)** +- Temporal client setup, four workers via `sientia_do.temporal.worker.prepare_worker`, plus a + fifth registered only when `IMPORT_MODEL_ENABLED` is truthy (`1`, `true`, `yes`, `on`) +- Runtime-scoped task queues: `{workflow}-{RUNTIME}-queue` for all workflows +- Metrics server initialization, notification handler setup +- Graceful shutdown and autoscaling-friendly behavior + +**Breaking (schedulers):** drift and simple_metrics queues are no longer `drift-queue` / +`simple_metrics-queue`. Use `drift-{RUNTIME}-queue` and `simple_metrics-{RUNTIME}-queue` +matching the worker pod `RUNTIME` env (same as `predictions_batch` / `minimal_retrain`). + +Task queues, one per worker: + +| Queue | Worker | Registered | +|---|---|---| +| `predictions_batch-{RUNTIME}-queue` | PredictionsBatch (+ its two sub-workflows) | always | +| `minimal_retrain-{RUNTIME}-queue` | MinimalRetrain | always | +| `drift-{RUNTIME}-queue` | Drift | always | +| `simple_metrics-{RUNTIME}-queue` | SimpleMetrics | always | +| `import_model-{RUNTIME}-queue` | ImportModel | only when `IMPORT_MODEL_ENABLED` is truthy | + +#### **Workflows (`laborious/workflows/`)** +- `predictions_batch.py`: Batch prediction entry point +- `sub_workflows/prediction_process.py`: Core prediction pipeline +- `sub_workflows/format_and_export_prediction.py`: Formatting and export +- `minimal_retrain.py`: Automated model retraining and production update +- `drift.py`: Data drift detection and monitoring +- `simple_metrics.py`: Regression metrics calculation (RMSE, MSE, MAE, RΒ²) +- `import_model.py`: `.sientia` bundle import β€” registered model + model document, behind + `IMPORT_MODEL_ENABLED` + +#### **Activities (`laborious/activities/`)** +- `gates.py`: Data quality validation, filtering, and data formatting operations + - Input/response/content gates for quality validation + - Prediction and transformed data formatting + - Retrain report formatting and metrics recording +- `mlflow.py`: Transform, predict, and model management operations + - MLFlow model transformation and prediction + - Model retraining and production updates + - Reference data retrieval from MLflow Model Registry +- `storage.py`: PostgreSQL queries and MinIO-aware data loading + - `load_query_with_minio_offload`: SQL load with automatic MinIO offload + - `export_payload_to_postgres`: Resolve MinIO payloads and export to Postgres + - `cleanup_minio_objects_expired`: Retention-based MinIO object cleanup + - `query_to_minio`: Legacy parquet upload for retraining data +- `model_metrics.py`: Drift detection and regression metrics + - Univariate and multivariate drift calculation + - Simple metrics (RMSE, MSE, MAE, RΒ²) +- `opc.py`: OPC UA export to industrial systems (optional) +- `api.py`: PI Web API export operations (optional) + - Prediction and confidence data writing to PI Web API + - Error handling and notification integration +- `activities.py`: Aggregates all activity interfaces (Storage, MLFlow, Gates, OPC, ModelMetrics, API) + +#### **Data Services (`laborious/utils/`)** +- `connectors_config.py`: Env-driven configuration builders +- `models/minio_dataframe_payload.py`: MinIO-offloaded DataFrame payload model +- `repository/model_repository.py`: MLFlow operations and retraining +- `repository/opc_repository.py`: OPC UA client, writes, session recovery (see [OPC UA Communication](#opc-ua-communication)) +- `repository/minio_manager.py`: MinIO object storage operations +- `filters/conditional_filters.py` and `filters/mlflow_filters.py` + +### Data Flow Architecture + +#### **1. Batch Prediction Pipeline** +``` +Input Data (PostgreSQL) β†’ Data Quality Gates β†’ MLFlow Transform β†’ +MLFlow Prediction β†’ Response Validation β†’ Format & Export + β”œβ”€β†’ Predictions β†’ PostgreSQL [+ OPC] [+ PI Web API] + └─→ Transformed Data β†’ PostgreSQL (optional) +``` + +#### **2. Model Retraining Pipeline** +``` +Training Data β†’ Model Retraining β†’ Quality Validation β†’ +Production Update β†’ Notification & Monitoring +``` + +#### **3. Drift Detection Pipeline** +``` +Target Data (PostgreSQL) + Reference Data (MLFlow) β†’ +Drift Calculation (univariate + multivariate) β†’ PostgreSQL Export +``` + +#### **4. Simple Metrics Pipeline** +``` +Predictions + Targets (PostgreSQL JOIN) β†’ +Metrics Calculation (RMSE, MSE, MAE, RΒ²) β†’ PostgreSQL Export +``` + +### Security Architecture + +#### **Authentication & Authorization** +- **MLFlow API Authentication**: Username/password +- **Database Security**: Encrypted connections and credential management +- **OPC Certificates** (if enabled): Client/server certs +- **PI Web API Authentication**: Bearer token or basic authentication +- **Kubernetes Secrets**: Secure secret storage + +#### **Network Security** +- TLS/SSL, network policies, service mesh, firewalls, VPN + +#### **Data Security** +- At-rest/in-transit encryption, RBAC, audit logging, 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"] + }, + "pi_web_api_output_config": { + "endpoint": "https://pi-server.com/piwebapi", + "prediction_tags": {"tag1": "web_id_1"}, + "confidence_tags": {"tag2": "web_id_2"} + } +} +``` + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. load_custom_query] --> B[2. prediction_process πŸ”ƒ] + + A -.-> Database[(Database)] +``` + +### 2. Prediction Process Workflow (`prediction_process.py`) + +The **PredictionProcess** workflow implements the core prediction pipeline for ML model inference. It handles data quality validation, MLFlow model interactions, and prediction processing. + +#### Purpose +- **Data Quality Validation**: Applies configurable filters for data integrity +- **MLFlow Integration**: Manages model transformation and prediction requests +- **Response Validation**: Filters MLFlow API responses for quality assurance +- **Prediction Export**: Delegates prediction formatting and export operations + +#### Execution Flow +1. **Input Data Gate**: Applies configured filters for data quality validation +2. **Path Decision**: Determines processing path based on filter results +3. **MLFlow Transform**: Requests data transformation using MLFlow models +4. **Response Validation**: Filters transform responses for quality assurance +5. **Content Validation**: Filters transformed data content for quality check +6. **MLFlow Prediction**: Executes prediction using transformed data +7. **Prediction Response Validation**: Filters prediction responses for final quality check +8. **Export Delegation**: Delegates to FormatAndExportPrediction workflow +9. **MinIO Cleanup**: Cleans up expired offloaded payloads (if any, in `finally` block) + +#### 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 +- **MinIO Cleanup**: Automatic retention-based cleanup of offloaded payloads +- **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": {...}, + "pi_web_api_output_config": { + "endpoint": "https://pi-server.com/piwebapi", + "prediction_tags": {"tag1": "web_id_1"}, + "confidence_tags": {"tag2": "web_id_2"} + } +} +``` + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. input_gate] --> B[2. request_transform] --> C[3. mlflow_response_gate] --> D[4. mlflow_content_gate] --> E[5. request_predict] --> F[6. mlflow_response_gate] --> G[7. format_and_export_predictionπŸ”ƒ] + G --> H[8. cleanup_minio_objects_expired] + + B -.-> MLFlow[MLFlow] + E -.-> MLFlow[MLFlow] + H -.-> MinIO[(MinIO)] +``` + +### 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 +- **PI Web API Integration**: Writes predictions and confidence to PI Web API for industrial systems +- **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. **Transformed Data Processing**: Optionally formats and exports transformed data separately +4. **PI Web API Export**: Writes predictions and confidence to PI Web API (if configured) +5. **OPC Export**: Writes predictions to OPC servers (if configured) +6. **PostgreSQL Export**: Writes formatted predictions to database +7. **Metrics Recording**: Records export performance and success metrics + +#### Key Features +- **Flexible Formatting**: Configurable output formats for different destinations +- **Multi-Destination Export**: PostgreSQL, OPC server, and PI Web API integration +- **Transformed Data Export**: Optional separate export of MLFlow transformed data +- **Performance Monitoring**: Comprehensive metrics for export operations +- **Error Handling**: Robust error handling with notification integration + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. format_prediction/format_default_prediction] --> B[2. format_transformed_data] --> C[3. write_pi_web_api_data] --> D[4. write_opc_data] --> E[5. export_data_to_postgres] --> F[6. write_metrics] + + A -.-> Format[Data Formatting] + B -.-> Transform[Transformed Data] + C -.-> PIWebAPI[PI Web API] + D -.-> OPC[OPC Servers] + E -.-> PostgreSQL[(PostgreSQL)] + F -.-> Prometheus[Prometheus] +``` + +#### Transformed Data Export +When `transformed_data` is provided in the input, the workflow will: +- Format the transformed data using `format_transformed_data` activity +- Export it to a separate table (`transform_table_name`) asynchronously +- Wait for both prediction and transformed data exports to complete +- This enables separate tracking of model transformations for analysis and debugging + +### 4. Minimal Retrain Workflow (`minimal_retrain.py`) + +The **MinimalRetrain** workflow handles automated model retraining and production model updates. + +#### Purpose +- **Model Retraining**: Automates ML model retraining processes +- **Production Updates**: Manages production model version updates +- **Data Export**: Exports training data for model development +- **Quality Assurance**: Ensures model quality before production deployment + +#### Execution Flow +1. **Data Loading**: Loads training data using custom queries +2. **Model Retraining**: Executes model retraining process +3. **Quality Validation**: Validates retrained model performance +4. **Production Update**: Updates production model if quality criteria met +5. **Data Export**: Exports training data for analysis + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. load_custom_query] --> B[2. retrain_model] --> C[3. update_production_model] --> D[4. export_data_to_postgres] + + A -.-> Database[(Database)] + B -.-> MLFlow[MLFlow] + C -.-> MLFlow[MLFlow] + D -.-> PostgreSQL[(PostgreSQL)] +``` + +### 5. Drift Workflow (`drift.py`) + +The **Drift** workflow detects data drift by comparing current data against a reference dataset from the MLflow Model Registry. + +#### Execution Flow +1. **Data Loading**: Loads target data and reference data in parallel +2. **Drift Calculation**: Calculates univariate and multivariate drift metrics +3. **Data Export**: Exports drift metrics to PostgreSQL + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. load_custom_query] --> C[3. calculate_drift] --> D[4. export_data_to_postgres] + B[2. get_reference_data] --> C + + A -.-> Database[(Database)] + B -.-> MLFlow[MLFlow] + D -.-> PostgreSQL[(PostgreSQL)] +``` + +#### Input Parameters +```json +{ + "schedule_name": "hourly_drift", + "model_name": "temperature_model", + "model_id": "temp_001", + "schema": "sientia_data", + "source_table_name": "laborious_data", + "target_table_name": "drift_metrics", + "interval": 60, + "model_config": { "target": "temperature" }, + "drift_metrics": ["kolmogorov_smirnov", "jensen_shannon", "wasserstein"], + "chunk_period": "min" +} +``` + +### 6. Simple Metrics Workflow (`simple_metrics.py`) + +The **SimpleMetrics** workflow calculates regression metrics (RMSE, MSE, MAE, RΒ²) by comparing predictions against actual target values. + +#### Execution Flow +1. **Data Loading**: Loads prediction vs target data via a JOIN query +2. **Metrics Calculation**: Calculates configured regression metrics +3. **Data Export**: Exports metrics to PostgreSQL + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. load_custom_query] --> B[2. calculate_simple_metrics] --> C[3. export_data_to_postgres] + + A -.-> Database[(Database)] + C -.-> PostgreSQL[(PostgreSQL)] +``` + +#### Input Parameters +```json +{ + "schedule_name": "hourly_metrics", + "model_name": "temperature_model", + "model_id": "temp_001", + "schema": "sientia_data", + "predictions_table_name": "predictions", + "data_table_name": "laborious_data", + "target_table_name": "simple_metrics", + "interval_minutes": 60, + "model_config": { "target": "temperature" }, + "metrics": ["rmse", "mse", "mae", "r2"] +} +``` + +### 7. Import Model Workflow (`import_model.py`) + +The **ImportModel** workflow turns an encrypted `.sientia` bundle already uploaded to MinIO into a +registered MLflow model plus one document in the MongoDB `models` collection β€” or into a durable +`ERROR` row carrying a sentence a person can read. It is registered **only** when +`IMPORT_MODEL_ENABLED` is set, so the importer can live in its own runtime with its own credentials. + +Full contract for the frontend and the BFF: [`docs/model-import.md`](docs/model-import.md). +The file format: [`docs/sientia-bundle-format.md`](docs/sientia-bundle-format.md). + +#### Execution Flow +1. **Claim**: takes ownership of the `public.experiment_run` row the BFF inserted (it never inserts one) +2. **Download**: object limits, then the mandatory `expected_digest` comparison +3. **Open**: seven ordered gates β€” header, decryption, archive inspection, extraction, structure, content policy +4. **Provision**: experiment, a new run with the recorded name, the artifact tree, a registered version (stage untouched) +5. **List**: the MongoDB model document, `active: false` +6. **Finish**: cleanup, then one terminal write classifying the outcome once + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. claim_import_status] --> B[2. download_import_bundle] --> C[3. open_import_bundle] + C --> D[4. create_import_experiment] --> E[5. upload_import_artifacts] + E --> F[6. register_import_model_version] --> G[7. write_import_model_document] + G --> H[8. cleanup_import_files] --> I[9. record_import_terminal_status] + + A -.-> BFF[(sientia-core-mlops-bff)] + B -.-> MinIO[(MinIO)] + F -.-> MLflow[(MLflow)] + G -.-> Mongo[(MongoDB)] + I -.-> BFF +``` + +#### Input Parameters +```json +{ + "import_run_id": 41, + "object_key": "imported_models/sales_forecast_v3.sientia", + "expected_digest": "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", + "password_envelope": "Base64OfNonceCiphertextTag==", + "bucket": "sientia" +} +``` + +A successful import leaves the registered version **unstaged** and the model document **inactive**: +promoting and activating are manual operator steps (QTZPOC-21). The `V12` migration of +`sientia-core-mlops-bff` must be applied before enabling the worker β€” see +[`docs/model-import.md`](docs/model-import.md) Β§ 7. + +## πŸ“‹ Prerequisites + +- Python 3.11+ +- Temporal server/cluster +- PostgreSQL database +- MLFlow server +- MinIO object storage (for MLFlow artifacts) +- MongoDB server (for notifications) +- OPC server(s) if using OPC export +- PI Web API server if using PI Web API export + +**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** + + 1. **Install github cli** + ```bash + sudo apt update + sudo apt install gh -y + ``` + + 2. **Authenticate with github** + ```bash + gh auth login + ``` + + 3. **Run the install_dependencies.sh script** + ```bash + chmod +x install_dependencies.sh + ./install_dependencies.sh + ``` + +4. **Create environment configuration file** + ```bash + cp .env.example .env + # Edit .env with your connection details + ``` + +5. **Configure external dependencies** + + You'll need to set up port forwarding or connections to external services. For example: + + ```bash + # Port forwarding from Kubernetes cluster + kubectl port-forward svc/postgresql 5432:5432 + kubectl port-forward svc/mlflow 5000:5000 + kubectl port-forward svc/mongodb 27017:27017 + + # Or connect to external services + # Ensure services are accessible on localhost with appropriate ports + ``` + +## πŸ“¦ How to Run + +### Running the Laborious Application + +Use the provided script to run the application locally: + +```bash +# Make script executable (first time only) +chmod +x scripts/run_local.sh + +# Run the application (from the repository root) +./scripts/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 scripts/run_coverage.sh + +# Run tests with coverage (from the repository root) +./scripts/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/laborious/activities/ +pytest tests/laborious/workflows/ +``` + +### 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 +``` + +### Running One Model Import by Hand + +[`scripts/import_model_manual_run.py`](scripts/import_model_manual_run.py) stands in for the frontend +that does not exist yet. The Angular screen plus the Spring Boot BFF will, between them, hash a +`.sientia` file, wrap the bundle password, upload the object and insert one `PENDING` row before +starting `import_model`; this script does exactly those four things against whatever `.env` points +at, then reads the outcome back and can undo itself. + +It is a **development tool**. It writes to whatever `.env` names β€” a real object in a real bucket and +a real row in a real `public.experiment_run` β€” and a `.env` copied from a cluster deployment can +perfectly well name production. It is not for production use. + +**Prerequisites** + +- A reachable environment: the BFF Postgres, MinIO, MongoDB, MLflow and Temporal that `.env` names. +- The `V12` migration applied to `public.experiment_run` in the BFF database. Without it the import + provisions the model and then cannot record the outcome β€” see + [`docs/model-import.md`](docs/model-import.md) Β§ 7. +- A worker running with `IMPORT_MODEL_ENABLED` truthy and the same `RUNTIME`, if you want cell 8 (or + the Temporal UI) to actually execute the run. +- The import variables from [`.env.example`](.env.example) filled in, including `IMPORT_PASSWORD_KEY` + and the script's own `IMPORT_SCRIPT_BUNDLE_PATH`, `IMPORT_SCRIPT_BUNDLE_PASSWORD` and + `IMPORT_SCRIPT_USERNAME`. + +**Running it.** The file is split into `# %%` cells: run them one at a time in the VS Code +interactive window, or run the module top to bottom. + +```bash +source ./venv/bin/activate + +# top to bottom (cells 5 and 6 write; cell 2 must pass first) +python scripts/import_model_manual_run.py +``` + +| Cell | Does | Writes? | +|---|---|---| +| 1 | Resolves and prints every target β€” Postgres, MinIO, Mongo, MLflow, Temporal, the task queue | no | +| 2 | **Confirmation guard** | no | +| 3 | Picks the bundle, computes `expected_digest`, prints the parsed header | no | +| 4 | Builds the AES-256-GCM password envelope | no | +| 5 | Uploads the object, re-reads it and checks the digest still matches | **yes** | +| 6 | Inserts the `PENDING` / `run_type='IMPORT'` row, prints `import_run_id` | **yes** | +| 7 | Prints the workflow input as JSON, ready for the Temporal UI | no | +| 8 | Optionally starts `import_model` and waits (`START_THE_WORKFLOW`) | starts a run | +| 9 | Reads the outcome back: the row, the MLflow experiment/run/version, the model document | no | +| 10 | Undo: deletes the object and the row always; the provisioning behind `DELETE_PROVISIONING` | deletes | + +**The guard.** Cells 5 and 6 are the only cells that write, and neither runs until cell 2 passes. +Cell 2 compares a confirmation value against the Postgres host cell 1 resolved: set +`IMPORT_SCRIPT_CONFIRM_TARGET` in `.env`, or paste the host into the cell. It is a read-then-restate +rather than a boolean on purpose β€” a flag can be flipped without ever looking at what it points at. + +No credential is written in the script and none is printed: it prints hosts, ports, database, bucket +and object names, the row id and the workflow input, and never the bundle password, the envelope key +or the envelope's plaintext. + +## Code Quality & Validation + +### Overview + +Since Python is not compiled, we validate quality, security, and correctness before execution. + +### Validation Tools + +- Ruff: Linting and formatting +- mypy: Static type checking +- Bandit: Security analysis +- pytest: Unit/integration testing with coverage + +### Tools Installation + +```bash +pip install -r requirements-dev.txt +``` + +### Complete Validation + +Run each validation step individually: +```bash +ruff format --check laborious/ tests/ +ruff check laborious/ tests/ +mypy laborious/ +bandit -r laborious/ -ll +pytest tests/ --cov=laborious --cov-report=term-missing +``` + +### Automatic Fixes + +```bash +ruff format laborious/ tests/ +ruff check --fix laborious/ tests/ +``` + +### Configuration + +All settings reside in `pyproject.toml` (Ruff, mypy, pytest, Bandit). + +### CI/CD Integration + +The workflow at `.github/workflows/quality-gate.yml` executes validations on each push/PR. + +### Best Practices + +- Run all validation steps before committing +- Use `ruff check --watch` for continuous feedback +- Add type hints and tests for new code + +## πŸ§ͺ Testing + +### Test Structure +``` +tests/ +β”œβ”€β”€ conftest.py # Global fixtures and env setup +β”œβ”€β”€ laborious/ +β”‚ β”œβ”€β”€ activities/ # Activity implementation tests +β”‚ β”‚ β”œβ”€β”€ test_activities.py # Activities aggregator tests +β”‚ β”‚ β”œβ”€β”€ test_gates.py # Data quality gates and formatting tests +β”‚ β”‚ β”œβ”€β”€ test_mlflow.py # MLFlow operations and reference data tests +β”‚ β”‚ β”œβ”€β”€ test_storage.py # Storage and MinIO offload tests +β”‚ β”‚ β”œβ”€β”€ test_model_metrics.py # Drift and simple metrics tests +β”‚ β”‚ β”œβ”€β”€ test_opc.py # OPC operations tests +β”‚ β”‚ └── test_api.py # PI Web API operations tests +β”‚ β”œβ”€β”€ workflows/ # Workflow orchestration tests +β”‚ β”‚ β”œβ”€β”€ test_predictions_batch.py +β”‚ β”‚ β”œβ”€β”€ test_minimal_retrain.py +β”‚ β”‚ β”œβ”€β”€ test_drift.py +β”‚ β”‚ β”œβ”€β”€ test_simple_metrics.py +β”‚ β”‚ └── subworkflows/ +β”‚ β”‚ β”œβ”€β”€ test_prediction_process.py +β”‚ β”‚ └── test_format_and_export_prediction.py +β”‚ └── utils/ # Utility function tests +β”‚ β”œβ”€β”€ test_connectors_config.py +β”‚ β”œβ”€β”€ models/ +β”‚ β”‚ └── test_minio_dataframe_payload.py +β”‚ β”œβ”€β”€ filters/ +β”‚ β”‚ β”œβ”€β”€ test_conditional_filters.py +β”‚ β”‚ └── test_mlflow_filters.py +β”‚ └── repository/ +β”‚ β”œβ”€β”€ test_model_repository.py +β”‚ └── test_opc_repository.py +``` + +### Test Coverage +The test suite provides comprehensive coverage for: +- **Data Quality Gates**: Input, response, and content validation filters +- **Data Formatting**: Prediction, transformed data, and retrain report formatting +- **MLFlow Operations**: Transform, predict, retrain, and reference data retrieval +- **Workflow Orchestration**: Complete workflow execution paths and error handling +- **Metrics Recording**: Performance monitoring and OPC export metrics + +### 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/laborious/activities/test_gates.py +pytest tests/laborious/workflows/test_predictions_batch.py +``` + +### End-to-End Tests (`e2e/`) + +The `e2e/` suite runs against real dependencies in Docker containers via +[testcontainers](https://testcontainers.com/): PostgreSQL, MinIO, MongoDB, and β€” for the model-import +suite β€” an `mlflow server` subprocess started from this repository's venv. It lives outside +`testpaths`, so it is never collected by the unit gate. + +**One-time setup, required before running anything under `e2e/`:** + +```bash +echo "ryuk.disabled=true" >> ~/.testcontainers.properties +``` + +Without it, **every** e2e test fails at setup with a Docker mount error and never reaches an +assertion: + +``` +docker.errors.APIError: 500 Server Error ... /containers//start: +Mounts denied: The path /socket_mnt/home//.docker/desktop/docker.sock +is not shared from the host and is not known to Docker. +``` + +The cause is not the tests. Before starting any test container, testcontainers starts a reaper +container ("Ryuk") that bind-mounts the Docker socket so it can clean up if the test process dies +without teardown. Under Docker Desktop on Linux that socket is `~/.docker/desktop/docker.sock`, a path +Docker Desktop's File Sharing does not expose, so the reaper's own container cannot start. Note the +symptom is `ERROR`, not `FAILED` β€” the failure is in fixture setup. + +Disabling it costs nothing in a normal run: every fixture stops its own containers in teardown. Ryuk +only matters when the process is killed outright (`kill -9`, a crash, a hard stop from the IDE). After +one of those, clean up with: + +```bash +docker ps -a --filter "label=org.testcontainers=true" -q | xargs -r docker rm -f +``` + +**Running it:** + +```bash +source ./venv/bin/activate + +# the predictions and OPC suites +pytest e2e -m "not import_e2e" + +# the .sientia model-import suite (opt-in; see e2e/scenarios.md Β§ 4) +pytest e2e -m import_e2e +``` + +Full scenario catalogue and the import suite's fidelity contract: +[`e2e/scenarios.md`](e2e/scenarios.md). + +## πŸ“Š Monitoring and Metrics + +The Laborious system exposes comprehensive Prometheus metrics for operational visibility and performance monitoring: + +### Application Health Metrics +- `app_up`: Application health status (1=healthy, 0=unhealthy) + - Labels: `pod_id` + +### Prediction Operation Metrics +- `laborious_predictions_written_count`: Counter for successful prediction exports + - Labels: `pod_id`, `model_name`, `workflow_name` +- `laborious_prediction_confidence_monitor`: Gauge for current prediction confidence levels + - Labels: `pod_id`, `model_name`, `workflow_name` +- `laborious_prediction_response_time_monitor`: Histogram for prediction response times + - Labels: `pod_id`, `model_name`, `workflow_name` + - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] + +### OPC Export Metrics +- `laborious_prediction_opc_writing_count`: Counter for OPC server write operations + - Labels: `pod_id`, `model_name`, `workflow_name`, `opc_server_id` +- `laborious_prediction_opc_writing_response_time_monitor`: Histogram for OPC write response times + - Labels: `pod_id`, `model_name`, `workflow_name`, `opc_server_id` + - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] + +OPC UA session and write diagnostics (Prometheus, `laborious/metrics.py`): + +- `opc_connections_initiated_total`, `opc_connections_failed_total`, `opc_connection_status` +- `opc_session_created_total`, `opc_session_closed_total`, `opc_session_revised_timeout_milliseconds` +- `opc_write_attempts_total` (label `result`: `OK` or exception name, e.g. `BadSessionIdInvalid`) +- `opc_write_inter_arrival_over_session_timeout_total` + +See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, and confidence codes **12** / **14**. + +### Data Quality Metrics +- Filter pass/fail rates through notification system +- MLFlow API response validation metrics +- Data quality gate performance tracking + +## βš™οΈ Configuration + +### Environment Variables + +| Variable | Description | Default | Required | +|----------|-------------|---------|----------| +| `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes | +| `TEMPORAL_NAMESPACE` | Temporal namespace | `laborious` | No | +| `RUNTIME` | Task queue suffix for all workflows (`{workflow}-{RUNTIME}-queue`) | _(none)_ | Yes | +| `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 | +| `POSTGRES_MIN_CONNECTIONS` | Minimum PostgreSQL connections | `5` | No | +| `POSTGRES_MAX_CONNECTIONS` | Maximum PostgreSQL connections | `20` | No | +| `MLFLOW_HOST` | MLFlow server hostname | `http://localhost` | Yes | +| `MLFLOW_PORT` | MLFlow server port | `5080` | Yes | +| `MLFLOW_USERNAME` | MLFlow username | `aignosi` | Yes | +| `MLFLOW_PASSWORD` | MLFlow password | `aignosi` | Yes | +| `IMPORT_MODEL_ENABLED` | Register the import worker on `import_model-{RUNTIME}-queue` | _(unset)_ | No | +| `IMPORT_PASSWORD_KEY` | base64 of the 32-byte AES-256-GCM key wrapping bundle passwords | _(none)_ | If importing | +| `IMPORT_BUNDLE_BUCKET` | Bucket holding uploaded `.sientia` objects | `sientia` | No | +| `IMPORT_BUNDLE_PREFIX` | Key prefix the uploads must sit under | `imported_models/` | No | +| `IMPORT_BUNDLE_RETENTION_DAYS` | Lifecycle expiry applied to that prefix | `7` | No | +| `IMPORT_WORK_DIR` | Worker scratch directory for bundles | `/sientia-import` | No | +| `IMPORT_MAX_OBJECT_BYTES` | Gate 1 object size ceiling | `1073741824` | No | +| `IMPORT_MAX_ARCHIVE_ENTRIES` | Gate 4 entry count ceiling | `5000` | No | +| `IMPORT_MAX_UNCOMPRESSED_BYTES` | Gate 4 uncompressed size ceiling | `4294967296` | No | +| `IMPORT_MAX_COMPRESSION_RATIO` | Gate 4 compression ratio ceiling | `200` | No | +| `IMPORT_MODELS_COLLECTION` | MongoDB collection holding the model listing | `models` | No | +| `IMPORT_STATUS_DB_NAME` | Import log database name (the BFF's, on the `POSTGRES_*` server) | `sientia-core-mlops-bff` | If importing | +| `IMPORT_STATUS_DB_HOST` | Override: import log host, when it is not the `POSTGRES_*` server | `POSTGRES_HOST` | No | +| `IMPORT_STATUS_DB_PORT` | Override: import log port | `POSTGRES_PORT` | No | +| `IMPORT_STATUS_DB_USER` | Override: import log user | `POSTGRES_USER` | No | +| `IMPORT_STATUS_DB_PASSWORD` | Override: import log password | `POSTGRES_PASSWORD` | No | +| `OPC_CONFIG` | OPC server configuration (JSON) | `{}` | No | +| `OPC_ID` | OPC server identifier | `1` | No | +| `OPC_URL` | OPC server URL | `opc.tcp://localhost:4840` | No | +| `OPC_SERVER_URI` | OPC server URI | `opc.tcp://localhost:4840` | No | +| `OPC_CERT_PATH` | OPC client certificate path | `None` | No | +| `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No | +| `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No | +| `OPC_RECONNECTION_INTERVAL` | Minimum seconds between OPC reconnects | `120` | No | +| `PI_WEB_API_BASE_URL` | PI Web API server base URL | `None` | No | +| `PI_WEB_API_AUTH_TYPE` | PI Web API authentication type (basic/bearer) | `None` | No | +| `PI_WEB_API_AUTH_TOKEN` | PI Web API authentication token | `None` | No | +| `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes | +| `MONGODB_USERNAME` | MongoDB username | `root` | Yes | +| `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes | +| `MONGODB_DATABASE_NAME` | MongoDB database name | `sientia` | Yes | +| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index hours | `1` | No | +| `LOG_LEVEL` | Application log level | `INFO` | No | +| `PROJECT_NAME` | Project name for metrics | `laborious` | No | +| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | +| `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | +| `POD_ID` | Kubernetes pod identifier | `None` | No | +| `MINIO_ENDPOINT_URL` | MinIO endpoint URL | `http://localhost:9000` | Yes | +| `MINIO_ACCESS_KEY` | MinIO access key | `minioadmin` | Yes | +| `MINIO_SECRET_KEY` | MinIO secret key | `minioadmin` | Yes | +| `MINIO_REGION_NAME` | MinIO region name | `us-east-1` | No | +| `MINIO_DEFAULT_BUCKET` | Default MinIO bucket | `laborious` | No | +| `MINIO_RETENTION_HOURS` | Retention window (hours) for offloaded MinIO objects | `24` | No | +| `SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES` | Offload threshold for DataFrame-derived payloads | `int(1.5 * 1024 * 1024)` | No | + + + + +### MinIO Payload Offload & Retention + +Laborious uses MinIO to prevent Temporal workflow history from carrying very large in-memory payloads (pandas `DataFrame`-derived dicts). +Whenever a payload exceeds a configurable size threshold, it is stored as a parquet file in MinIO and the workflow history only keeps a lightweight reference. + +Notes: +- `SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES` supports: + - Integer bytes (e.g. `"1572864"`) + - Float MiB (e.g. `"1.5"`), converted to bytes as `MiB * 1024 * 1024` +- Fallback behavior uses `1.5 MiB` when the env var is missing or invalid. + +#### Wire Contract: `MinioDataFramePayload` + +The payload is implemented in `laborious/utils/models/minio_dataframe_payload.py`. +The dataclass does **not** store a pandas `DataFrame` field. +Instead, the `DataFrame` is only used at build time by: +- `MinioDataFramePayload.from_dataframe(...)` +- `MinioDataFramePayload.from_dataframe_to_dict(...)` + +After evaluation, the payload is serialized for Temporal as a flat dict: +- **Inline path**: `data` contains `df.to_dict()`, and MinIO keys (`object_key`, `bucket`, ...) are absent / `None`. +- **MinIO path**: the dict contains: + - `bucket` + - `object_key` (full MinIO object name returned by `MinioRepository.upload_file`) + - `object_prefix` (directory prefix used for cleanup listing; relative to the repository namespace) + - `uri` (best-effort `s3:///<...>` string) + - `data` is omitted / set to `None`. + +When an activity needs pandas operations, it resolves references using: +- `MinioDataFramePayload.retrieve(minio_repo)` β€” downloads from MinIO or returns inline data as a DataFrame + +#### MinIO Object Naming (Retention Parsing) + +MinIO object basename (required convention): +`{model_name}-{operation}-{timestamp}.parquet` + +Where: +- `model_name`: model identifier used by the pipeline +- `operation`: `initial` (SQL/query load before transform) or `transform` (after MLFlow transform) +- `timestamp`: `DATETIME_FORMAT_FILENAME` from `sientia_do.temporal.constants` + +The relative object key (under the repository namespace) is always shaped as: +`training_datasets/{model_name}/{basename}` + +Retention cleanup parses timestamps from the basename using the `-initial-` / `-transform-` anchors. +`model_name` may contain hyphens; parsing is resilient to it. + +#### Workflows / Activities Integration + +Predictions batch uses MinIO offload as follows: +1. `predictions_batch` calls `Activities.load_query_with_minio_offload` + - On success, it puts the serialized `MinioDataFramePayload` dict into `prediction_input["data"]`. +2. `sub_workflows/prediction_process` + - Tracks which MinIO prefixes were referenced for offloaded payloads. + - Runs `Activities.cleanup_minio_objects_expired` in a `finally` block (only when MinIO offload happened). +3. `laborious/activities/gates.py` and `laborious/activities/mlflow.py` + - Resolve offloaded payloads transparently before constructing pandas `DataFrame` objects. + +#### Legacy: `query_to_minio` (Minimal Retrain) + +`Storage.query_to_minio` is intentionally kept with its legacy behavior for `minimal_retrain`. +It always uploads parquet and returns `{success, object_key, uri}`. +It is not used by predictions batch MinIO offload, and its objects are not part of the retention parser described above. + +Legacy MinIO object layout (relative key): +`training_datasets/{model_name}/{object_prefix}_{timestamp}.parquet` where `object_prefix` is sanitized +(slashes replaced by underscores) to keep a stable model-level directory. + +## Model Flavors + +`model_config.predict_flavor` / `model_config.transform_flavor` accept `sklearn`, `pyfunc`, +`pytorch` and `joblib`. The flavor is configuration, never discovered from the artifact, and there is +no fallback between flavors: a failing loader propagates. + +`joblib` reads the artifact directly β€” download, first top-level `.pkl`, `joblib.load` β€” with the +artifact's `code/` put on `sys.path` first, and it writes on retrain (`model.pkl` + `code/`, no +`MLmodel`, so only this runtime reads it back). + +Full reference, including how to switch an existing model over: **[docs/model-flavors.md](docs/model-flavors.md)**. + +## OPC UA Communication + +Full reference: **[docs/opc-communication.md](docs/opc-communication.md)** (connection lifecycle, Tier-1 `Bad*` reconnect, connection lock / session readiness, metrics, PostgreSQL confidence **12** vs **14**, tests). + +Implementation plan: [`.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md`](.cursor/plans/opc_bad_reconnect_ac4c6045.plan.md). + +### 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` +- `OPC_CERT_PATH` +- `OPC_PRIVATE_KEY_PATH` +- `OPC_SERVER_CERT_PATH` +- `OPC_RECONNECTION_INTERVAL` + +### PI Web API Configuration + +PI Web API configuration is built from environment variables using the `build_api_config` function from `sientia_do.connectors_config`. The configuration includes: + +- `PI_WEB_API_BASE_URL`: Base URL of the PI Web API server +- `PI_WEB_API_AUTH_TYPE`: Authentication type ('basic' or 'bearer') +- `PI_WEB_API_AUTH_TOKEN`: Authentication token for API access + +The PI Web API export is optional and can be configured per workflow through the `pi_web_api_output_config` parameter: + +```json +{ + "pi_web_api_output_config": { + "endpoint": "https://pi-server.com/piwebapi", + "prediction_tags": { + "tag1": "web_id_1", + "tag2": "web_id_2" + }, + "confidence_tags": { + "tag3": "web_id_3", + "tag4": "web_id_4" + } + } +} +``` + +Where: +- `endpoint`: PI Web API endpoint URL +- `prediction_tags`: Dictionary mapping tag names to web IDs for prediction values +- `confidence_tags`: Dictionary mapping tag names to web IDs for confidence values + +### Workflow Configuration + +MongoDB pipeline configuration: + +#### Predictions Batch Workflow configuration sample + +This is the configuration for the Predictions Batch Workflow, to be inserted into the MongoDB pipeline collection. + +```json +{ + "schedule_name": "laborious-orchestrated-pipeline", + "model_id": "1", + "workflow_type": "predictions_batch", + "frequency": "30s", + "max_retry_policy": 1, + "query": "select * from sientia_data.laborious_data where model_id = 1 and \"timestamp\" > NOW() - INTERVAL '5 minutes' order by \"timestamp\" desc limit 30;", + "write_tags": [ + { + "server_id": "server1", + "type": "prediction", + "addr": "ns=2;i=5", + "data_type": "double" + }, + { + "server_id": "server1", + "type": "confidence", + "addr": "ns=2;i=6", + "data_type": "double" + } + ], + "input_filters": { + "EMPTY_DATA": {"POLICY": "STOP"}, + "SPECIFIC_VARIABLES_NULL_VALUES": { + "POLICY": "CONTINUE", + "config": {"variables": ["Counter"]} + } + }, + "mlflow_transform_filters": { + "API_ERROR": {"POLICY": "REPEAT"}, + "NAN_VALUES": {"POLICY": "STOP"} + }, + "mlflow_predict_filters": { + "API_ERROR": {"POLICY": "CONTINUE"} + }, + "path_priority": ["STOP", "CONTINUE", "REPEAT"], + "active": true, + "updated_at": { + "$date": "2025-09-16T10:00:00.000Z" + }, + "datetime_columns": ["timestamp", "created_at"], + "predictions_storage_policy": "lts:1" +} +``` + +This is the configuration created by the Orchestrator in Temporal. + +```json +{ + "datetime_columns":["timestamp","created_at"], + "frequency":"15m", + "input_filters":{"EMPTY_DATA":{"config":{},"policy":"STOP"}}, + "max_retry_policy":1, + "mlflow_predict_filters":{"API_ERROR":{"config":{},"policy":"CONTINUE"}}, + "mlflow_transform_filters":{ + "API_ERROR":{"config":{},"policy":"CONTINUE"}, + "EMPTY_DATA":{"config":{},"policy":"STOP"} + }, + "model_config":{ + "is_compressed":true, + "predict_flavor":"pyfunc", + "retention_minutes":60, + "retention_target":"artifact", + "transform_function_keyword":"transform" + }, + "model_id":"352", + "model_name":"courier", + "opc_output_config":{}, + "path_priority":["STOP","CONTINUE","REPEAT"], + "predictions_storage_policy":"lts:1", + "query":"select * from sientia_data.laborious_data where model_id = 352 order by \"timestamp\" desc limit 300;", + "retention_time":3600, + "schedule_name":"laborious-courier", + "schema":"sientia_data", + "table_name":"predictions", + "updated_at":"2025-09-12 19:35:01.600000+0000", + "workflow_type":"predictions_batch" +} +``` + +## πŸ”§ Development + +### Project Structure +``` +laborious/ +β”œβ”€β”€ activities/ # Temporal activity implementations +β”‚ β”œβ”€β”€ activities.py # Main activities aggregator +β”‚ β”œβ”€β”€ gates.py # Data quality gates and filtering +β”‚ β”œβ”€β”€ mlflow.py # MLFlow model operations +β”‚ β”œβ”€β”€ storage.py # PostgreSQL queries and MinIO offload +β”‚ β”œβ”€β”€ model_metrics.py # Drift and regression metrics +β”‚ β”œβ”€β”€ opc.py # OPC server operations +β”‚ └── api.py # PI Web API operations +β”œβ”€β”€ workflows/ # Temporal workflow definitions +β”‚ β”œβ”€β”€ predictions_batch.py # Main batch prediction workflow +β”‚ β”œβ”€β”€ minimal_retrain.py # Model retraining workflow +β”‚ β”œβ”€β”€ drift.py # Data drift detection workflow +β”‚ β”œβ”€β”€ simple_metrics.py # Regression metrics 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 (uses sientia_do prepare_worker) +β”œβ”€β”€ utils/ # Utility functions +β”‚ β”œβ”€β”€ connectors_config.py # Environment-driven config builders +β”‚ β”œβ”€β”€ models/ # Data models +β”‚ β”‚ └── minio_dataframe_payload.py # MinIO-offloaded DataFrame payload +β”‚ β”œβ”€β”€ 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 +β”‚ └── minio_manager.py # MinIO object storage 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** + - See [docs/opc-communication.md](docs/opc-communication.md) + - Verify OPC server is accessible and `OPC_RECONNECTION_INTERVAL` is appropriate + - Check certificate and key file paths + - Correlate `opc_write_attempts_total` with `opc_session_*` metrics; count session errors via `prediction_confidence = 14` + - Review OPC server logs for connection issues + +5. **PI Web API Connection Failures** + - Verify PI Web API server is accessible + - Check authentication credentials and token validity + - Verify web IDs exist and have write permissions + - Review PI Web API server logs for connection issues + - Check notification system for error details + +6. **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 +export LOG_LEVEL=DEBUG +``` + +## ⚑ 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**: One worker pod per `RUNTIME`; queues are + `predictions_batch-{RUNTIME}-queue`, `minimal_retrain-{RUNTIME}-queue`, + `drift-{RUNTIME}-queue`, `simple_metrics-{RUNTIME}-queue` +- **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.