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

@@ -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
+4. **PI Web API Export**: Writes predictions and confidence to PI Web API (if configured)
-5. **OPC Export**: Writes predictions to OPC servers
+5. **OPC Export**: Writes predictions to OPC servers (if configured)
-6. **Metrics Recording**: Records export performance and success metrics
+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]
+    C -.-> PIWebAPI[PI Web API]
-    D -.-> PostgreSQL[(PostgreSQL)]
+    D -.-> OPC[OPC Servers]
-    E -.-> Prometheus[Prometheus]
+    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

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

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:
+            input_data (dict[str, Any]): The input data containing:
-                - metadata (dict[str, Any]): The metadata.
+                - metadata (dict[str, Any]): Workflow execution metadata
-                - pi_web_api_output_config (dict[str, Any]): The PI Web API output configuration.
+                - pi_web_api_output_config (dict[str, Any]): PI Web API configuration with:
-                - data (dict[str, Any]): The data to write.
+                    - 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'])

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

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

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
+        3. Exporting data to PI Web API for real-time industrial access (if configured)
-        4. Persisting data to PostgreSQL database with comprehensive metadata
+        4. Exporting data to OPC servers for real-time industrial access (if configured)
-        5. Recording performance metrics for operational monitoring
+        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

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']