From 7cfa34a963906055faff3ce4a8a23b9e0d1d93ba Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 9 Jan 2026 09:30:26 -0300 Subject: [PATCH] SIENTIAPDE-1478 Enhance README and Codebase with PI Web API Integration - Updated README.md to include details about PI Web API integration, including configuration and export capabilities. - Modified Activities class to incorporate PI Web API export operations and error handling. - Added new API class for handling PI Web API interactions, including writing prediction and confidence data. - Updated prediction workflows to support PI Web API output configuration. - Enhanced worker and sub-workflows to include PI Web API in task queues and export processes. - Improved documentation and error handling for PI Web API connections and configurations. --- README.md | 88 ++++++++++++++++--- laborious/activities/activities.py | 7 +- laborious/activities/api.py | 31 +++++-- laborious/worker/worker.py | 1 + laborious/workflows/predictions_batch.py | 3 + .../format_and_export_prediction.py | 12 ++- .../sub_workflows/prediction_process.py | 19 +++- 7 files changed, 132 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index a34b939..8a7ac46 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # 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 automated retraining with strong data quality validation and observability. +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 @@ -70,7 +70,7 @@ A comprehensive, Temporal-based ML orchestration system for industrial data proc - **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 and OPC server integration for industrial systems +- **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 @@ -139,6 +139,9 @@ Laborious uses a Temporal-based architecture with strong separation of concerns - Model retraining and production updates - Reference data retrieval from MLflow Model Registry - `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 activity interfaces #### **Data Services (`laborious/utils/`)** @@ -153,7 +156,7 @@ Laborious uses a Temporal-based architecture with strong separation of concerns ``` Input Data (PostgreSQL) → Data Quality Gates → MLFlow Transform → MLFlow Prediction → Response Validation → Format & Export - ├─→ Predictions → PostgreSQL [+ OPC] + ├─→ Predictions → PostgreSQL [+ OPC] [+ PI Web API] └─→ Transformed Data → PostgreSQL (optional) ``` @@ -169,6 +172,7 @@ Production Update → Notification & Monitoring - **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** @@ -229,6 +233,11 @@ The **PredictionsBatch** workflow is the main entry point for batch prediction p "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"} } } ``` @@ -298,7 +307,12 @@ The **PredictionProcess** workflow implements the core prediction pipeline for M }, "model_retention": 60, "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "opc_output_config": {...} + "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"} + } } ``` @@ -321,19 +335,21 @@ The **FormatAndExportPrediction** workflow handles prediction data formatting an - **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. **PostgreSQL Export**: Writes formatted predictions to database -5. **OPC Export**: Writes predictions to OPC servers -6. **Metrics Recording**: Records export performance and success metrics +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 and OPC server integration +- **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 @@ -341,13 +357,14 @@ The **FormatAndExportPrediction** workflow handles prediction data formatting an #### Architecture Diagram ```mermaid flowchart LR - A[1. format_prediction/format_default_prediction] --> B[2. format_transformed_data] --> C[3. write_opc_data] --> D[4. export_data_to_postgres] --> E[5. write_metrics] + 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 -.-> OPC[OPC Servers] - D -.-> PostgreSQL[(PostgreSQL)] - E -.-> Prometheus[Prometheus] + C -.-> PIWebAPI[PI Web API] + D -.-> OPC[OPC Servers] + E -.-> PostgreSQL[(PostgreSQL)] + F -.-> Prometheus[Prometheus] ``` #### Transformed Data Export @@ -394,6 +411,7 @@ flowchart LR - 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 @@ -684,6 +702,9 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi | `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No | | `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No | | `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No | +| `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 | @@ -734,6 +755,37 @@ For single OPC server, use individual environment variables: - `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: @@ -832,7 +884,8 @@ laborious/ │ ├── activities.py # Main activities orchestrator │ ├── gates.py # Data quality gates and filtering │ ├── mlflow.py # MLFlow model operations -│ └── opc.py # OPC server operations +│ ├── 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 @@ -885,7 +938,14 @@ laborious/ - Check certificate and key file paths - Review OPC server logs for connection issues -5. **Workflow Execution Failures** +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 diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index bd8553d..f009b10 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -24,15 +24,18 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API): 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 + - Storage: 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 + - ModelMetrics: Model performance metrics and drift detection + - API: PI Web API export operations for industrial systems Attributes: postgres_config (dict): PostgreSQL connection configuration mlflow_config (dict): MLFlow server configuration opc_config (dict): OPC server configuration + pi_web_api_config (dict): PI Web API server configuration logger (Logger): Logging and observability instance notification_handler (NotificationHandler): Notification management instance """ @@ -137,6 +140,8 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API): This method ensures proper cleanup of all resources including: - PostgreSQL connection pools - OPC server connections + - PI Web API client connections + - MLFlow model repositories - Any other resources that need explicit cleanup The method should be called before the application terminates to ensure diff --git a/laborious/activities/api.py b/laborious/activities/api.py index 50be4f0..be8ffb3 100644 --- a/laborious/activities/api.py +++ b/laborious/activities/api.py @@ -18,10 +18,16 @@ PI_WEB_API_PREDICTION_ERROR_CONFIDENCE = 13 class API(SientiaMonitoring): """ - PI Web API operations for writing data to PI Web API. + PI Web API operations for writing prediction data to PI Web API. This class provides Temporal activities for interacting with the PI Web API - to write data to PI Web API. + to write prediction and confidence values to industrial systems. It handles + error scenarios gracefully by setting error confidence values and sending + notifications when write operations fail. + + The class implements comprehensive error handling for both prediction and + confidence value writes, ensuring that partial failures are properly + reported and handled. """ def __init__( @@ -66,13 +72,24 @@ class API(SientiaMonitoring): @activity.defn(name='write_pi_web_api_data') async def write_pi_web_api_data(self, input_data: dict[str, Any]) -> dict[Any, Any]: """ - Write data to PI Web API. + Write prediction and confidence data to PI Web API. + + This method writes prediction values and confidence scores to PI Web API + using configured web IDs. It handles errors gracefully by setting error + confidence values when prediction writes fail and sending notifications + for both prediction and confidence write errors. Args: - input_data (dict[str, Any]): The input data. Containing: - - metadata (dict[str, Any]): The metadata. - - pi_web_api_output_config (dict[str, Any]): The PI Web API output configuration. - - data (dict[str, Any]): The data to write. + input_data (dict[str, Any]): The input data containing: + - metadata (dict[str, Any]): Workflow execution metadata + - pi_web_api_output_config (dict[str, Any]): PI Web API configuration with: + - endpoint (str): PI Web API endpoint URL + - prediction_tags (dict[str, str]): Mapping of tag names to web IDs for predictions + - confidence_tags (dict[str, str]): Mapping of tag names to web IDs for confidence + - data (dict[str, Any]): Prediction data, its a dataframe converted to dict. + Returns: + dict[Any, Any]: Data dictionary with potentially modified confidence values + If prediction write fails, prediction_confidence is set to error value (13) """ metadata = input_data['metadata'] data = DataFrame(input_data['data']) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 60990cb..c6248df 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -7,6 +7,7 @@ prediction and retraining workflows. The worker supports multiple task queues: - predictions_batch-queue: Handles batch prediction workflows (heavy workload) + Includes activities for MLFlow, data quality gates, OPC export, PI Web API export, and PostgreSQL - minimal_retrain-queue: Handles model retraining workflows - drift-queue: Handles drift detection workflows - simple_metrics-queue: Handles simple metrics calculation workflows diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index 40bef4e..42be048 100644 --- a/laborious/workflows/predictions_batch.py +++ b/laborious/workflows/predictions_batch.py @@ -61,7 +61,10 @@ class PredictionsBatch: - 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 + - pi_web_api_output_config (dict, optional): PI Web API export configuration - datetime_columns (list[str], optional): Columns to treat as datetime + - save_transform (bool, optional): Whether to save transformed data (default: True) + - prediction_store_policy (str, optional): Data retention policy (default: 'lts:1') Returns: None: The workflow completes successfully when the child workflow finishes diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index bde9ff3..6663f61 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/laborious/workflows/sub_workflows/format_and_export_prediction.py @@ -26,6 +26,7 @@ class FormatAndExportPrediction: Export Destinations: - PostgreSQL Database: Persistent storage with timestamp conversion + - PI Web API: Real-time industrial system integration for prediction and confidence values - OPC Servers: Real-time industrial system integration - Prometheus Metrics: Performance monitoring and operational visibility """ @@ -38,9 +39,10 @@ class FormatAndExportPrediction: 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 + 3. Exporting data to PI Web API for real-time industrial access (if configured) + 4. Exporting data to OPC servers for real-time industrial access (if configured) + 5. Persisting data to PostgreSQL database with comprehensive metadata + 6. Recording performance metrics for operational monitoring The method implements flexible formatting strategies: - Normal predictions: Full data formatting with confidence scores @@ -61,8 +63,10 @@ class FormatAndExportPrediction: - model_name (str): Name of the ML model - 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 Optional keys: + - opc_output_config (dict[str, Any]): OPC server export configuration + - pi_web_api_output_config (dict[str, Any]): PI Web API export configuration + Contains endpoint, prediction_tags, and confidence_tags mappings - transformed_data (dict[str, Any]): Transformed data to export separately Only processed when path_flag is None - transform_table_name (str): Target table for transformed data export diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 88e9e3c..991d262 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -66,7 +66,10 @@ class PredictionProcess: - 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 + - opc_output_config (dict, optional): OPC server export configuration + - pi_web_api_output_config (dict, optional): PI Web API export configuration + - save_transform (bool, optional): Whether to save transformed data (default: True) + - prediction_store_policy (str, optional): Data retention policy (default: 'lts:1') Returns: None: The workflow completes successfully when export workflow finishes @@ -235,7 +238,17 @@ class PredictionProcess: Args: data: Input data for processing path_flag: Path decision from filter (STOP, CONTINUE, REPEAT) - input_data: Complete workflow input configuration + input_data: Complete workflow input configuration including: + - metadata (dict): Workflow execution metadata + - schema (str): Database schema + - table_name (str): Target table for predictions + - transform_table_name (str): Target table for transformed data + - model_id (str): ML model identifier + - model_name (str): ML model name + - model_config (dict, optional): Model configuration + - opc_output_config (dict, optional): OPC server export configuration + - pi_web_api_output_config (dict, optional): PI Web API export configuration + - prediction_store_policy (str, optional): Data retention policy confidence: Confidence level from filter validation last_timestamp: Last processed timestamp comment: Additional information about the filter result @@ -245,7 +258,7 @@ class PredictionProcess: Path Handling: - STOP: Terminates workflow execution - - CONTINUE: Proceeds with normal processing + - CONTINUE: Delegates to FormatAndExportPrediction workflow with current data - REPEAT: Repeats last prediction if available """ metadata = input_data['metadata']