SIENTIAPDE-1478
Enhance PI Web API Scouter Workflow and Documentation - Added support for PI Web API data ingestion, including real-time and historical data retrieval. - Implemented timestamp normalization to ensure consistency across records. - Updated README.md to reflect new features and detailed workflow execution flow. - Enhanced API class with comprehensive error handling and cleanup operations. - Improved CoreScouter workflow with early exit conditions and detailed processing steps. - Updated configuration parameters for better clarity and usability.
This commit is contained in:
vitor-aignosi
112
README.md
112
README.md
@@ -5,7 +5,7 @@ A high-performance, scalable data processing and ML model orchestration system b
|
||||
## Features
|
||||
|
||||
### Core Functionality
|
||||
- **Multi-Source Data Ingestion**: Support for Kafka topics, direct OPC server access, and real-time triggers
|
||||
- **Multi-Source Data Ingestion**: Support for Kafka topics, direct OPC server access, PI Web API endpoints, and real-time triggers
|
||||
- **Temporal Workflow Orchestration**: Robust workflow management with automatic retry policies and fault tolerance
|
||||
- **Data Quality Gates**: Configurable filtering for null values, out-of-bounds data, and custom validation rules
|
||||
- **Time-Series Aggregation**: Flexible aggregation functions (average, median, max, min, latest) with configurable parameters
|
||||
@@ -102,9 +102,14 @@ The **CoreScouter** workflow implements the core data processing pipeline for in
|
||||
1. **Data Quality Gate**: Applies configured filters (null values, out-of-bounds, custom rules)
|
||||
2. **Data Aggregation**: Groups data by tag and name, applies aggregation functions
|
||||
3. **Data Grouping**: Organizes data and stores temporarily in Redis with TTL
|
||||
4. **Data Export**: Persists processed data to PostgreSQL database
|
||||
4. **Data Export**: Persists processed data to PostgreSQL database with timestamp conversion
|
||||
5. **Metrics Recording**: Writes processing metrics for operational visibility
|
||||
|
||||
**Note**: The data export step uses timestamp conversion to ensure consistent datetime
|
||||
formatting. The export operation receives the schema, table name, data, and timestamp
|
||||
conversion configuration. Conflict resolution and unique column constraints are handled
|
||||
by the underlying PostgreSQL activity implementation.
|
||||
|
||||
#### Aggregation Functions
|
||||
- **`lts`**: Latest value (most recent data point)
|
||||
- **`avg`**: Average of all values in the group
|
||||
@@ -171,6 +176,102 @@ When `debug_data_package` is set to `true`, the workflow stores both raw and pro
|
||||
- Validating data transformations
|
||||
- Auditing data quality gate decisions
|
||||
|
||||
|
||||
### 3. PI Web API Scouter Workflow (`pi_web_api_scouter.py`)
|
||||
|
||||
The **PI Web API Scouter** workflow serves as the entry point for PI Web API data processing pipelines. Unlike the standard Scouter workflow that loads data from MongoDB collections, this workflow directly queries PI Web API endpoints to retrieve tag values and processes them for downstream use.
|
||||
|
||||
#### Purpose
|
||||
- **Direct API Ingestion**: Retrieves data directly from PI Web API endpoints
|
||||
- **Real-time Data Processing**: Supports real-time and historical data retrieval
|
||||
- **Data Normalization**: Normalizes timestamps to ensure consistency across records
|
||||
- **Workflow Orchestration**: Delegates data processing to the CoreScouter workflow
|
||||
- **Error Handling**: Comprehensive error handling with retry policies
|
||||
|
||||
#### Execution Flow
|
||||
1. **Tag Value Retrieval**: Retrieves tag values from PI Web API using configured WebIds and time periods
|
||||
2. **Data Normalization**: Normalizes timestamps to ensure all records in a batch share the same timestamp value
|
||||
3. **Data Validation**: Validates retrieved data and handles empty responses
|
||||
4. **Data Processing**: Delegates data processing to the CoreScouter child workflow
|
||||
|
||||
**Note**: The timestamp normalization process converts all timestamps to string format and then sets all records to the maximum timestamp value (lexicographically) found in the dataset. This ensures consistency across all records in a single batch.
|
||||
|
||||
#### Key Features
|
||||
- **Configurable Time Periods**: Supports flexible time period configurations (e.g., '*-1d', '*-1h')
|
||||
- **Data Point Limits**: Configurable maximum data points per tag via `max_count` parameter
|
||||
- **Timeout Management**: Configurable API request timeouts for reliable operation
|
||||
- **Empty Data Handling**: Gracefully handles empty responses without processing
|
||||
- **Standardized Processing**: Uses CoreScouter for consistent data quality and export operations
|
||||
|
||||
#### Input Parameters
|
||||
```json
|
||||
{
|
||||
"model_name": "pi_sensors",
|
||||
"model_id": "pi_001",
|
||||
"schedule_name": "hourly_pi_collection",
|
||||
"pi_web_api_query": {
|
||||
"endpoint": "/streamsets/recorded",
|
||||
"period": "*-1d",
|
||||
"max_count": 10,
|
||||
"api_timeout": 30
|
||||
},
|
||||
"model_tags": {
|
||||
"Temperature": {
|
||||
"webid": "F1AbCdEfGhIjKlMnOpQrStUvWxYz",
|
||||
"aggr_function": "avg",
|
||||
"data_range": [-50, 150]
|
||||
},
|
||||
"Pressure": {
|
||||
"webid": "F2AbCdEfGhIjKlMnOpQrStUvWxYz",
|
||||
"aggr_function": "max",
|
||||
"data_range": [0, 100]
|
||||
}
|
||||
},
|
||||
"trigger_laborious": false,
|
||||
"filters": {
|
||||
"OUT_OF_BOUNDS_FILTER": {"policy": "DISCARD"},
|
||||
"NULL_VALUES_FILTER": {"policy": "DISCARD"}
|
||||
},
|
||||
"schema": "sensor_data",
|
||||
"table_name": "pi_readings",
|
||||
"retention_time": 3600,
|
||||
"fill_missing_tags": false,
|
||||
"debug_data_package": false
|
||||
}
|
||||
```
|
||||
|
||||
**PI Web API Query Parameters:**
|
||||
- `endpoint` (str): PI Web API endpoint path (e.g., '/streamsets/recorded')
|
||||
- `period` (str): Time period configuration (e.g., '*-1d' for last day, '*-1h' for last hour)
|
||||
- `max_count` (int, optional): Maximum data points per tag. Defaults to 1
|
||||
- `api_timeout` (int): Request timeout in seconds for PI Web API calls
|
||||
|
||||
**Model Tags Configuration:**
|
||||
- `webid` (str): PI Web API WebId for the tag
|
||||
- `aggr_function` (str): Aggregation method (avg, mdn, max, min, lts)
|
||||
- `data_range` (list[int]): [min, max] values for data validation
|
||||
|
||||
#### Architecture
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
A[1. get_tag_values] --> B{data empty?}
|
||||
B -->|yes| C[Exit]
|
||||
B -->|no| D[2. core_scouter 🔃]
|
||||
|
||||
A -.-> PI_API[(PI Web API)]
|
||||
D -.-> CoreScouter[CoreScouter Workflow]
|
||||
```
|
||||
|
||||
#### Data Normalization
|
||||
|
||||
The `get_tag_values` activity normalizes timestamps to ensure consistency:
|
||||
1. Converts all timestamps to string format using the configured datetime format
|
||||
2. Identifies the maximum timestamp value (lexicographically) in the dataset
|
||||
3. Sets all records to use this normalized timestamp value
|
||||
|
||||
This normalization ensures that all records in a single batch share the same timestamp, which is useful for batch processing and data consistency in downstream operations.
|
||||
|
||||
## 📋 Prerequisites
|
||||
|
||||
- Python 3.11+
|
||||
@@ -179,6 +280,7 @@ When `debug_data_package` is set to `true`, the workflow stores both raw and pro
|
||||
- Redis server
|
||||
- MongoDB server
|
||||
- Kafka cluster (for data ingestion)
|
||||
- PI Web API server (for PI Web API Scouter workflow)
|
||||
|
||||
**Note**: External dependencies must be available either through:
|
||||
- Kubernetes cluster deployment
|
||||
@@ -365,6 +467,9 @@ The Scouter system exposes comprehensive Prometheus metrics:
|
||||
| `REDIS_PORT` | Redis port | `6379` | Yes |
|
||||
| `MONGODB_URL` | MongoDB connection URI | `localhost:27017` | Yes |
|
||||
| `KAFKA_BOOTSTRAP_SERVERS` | Kafka broker addresses | `localhost:9092` | No |
|
||||
| `PI_WEB_API_BASE_URL` | PI Web API base URL | - | Yes (for PI Web API Scouter) |
|
||||
| `PI_WEB_API_AUTH_TYPE` | PI Web API authentication type ('basic' or 'bearer') | - | Yes (for PI Web API Scouter) |
|
||||
| `PI_WEB_API_AUTH_TOKEN` | PI Web API authentication token | - | Yes (for PI Web API Scouter) |
|
||||
| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
|
||||
| `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No |
|
||||
| `PROJECT_NAME` | Project identifier for notifications | `scouter` | No |
|
||||
@@ -460,11 +565,13 @@ MongoDB pipeline configuration:
|
||||
scouter/
|
||||
├── activities/ # Temporal activity implementations
|
||||
│ ├── activities.py # Main activities orchestrator
|
||||
│ ├── api.py # PI Web API operations (tag value retrieval)
|
||||
│ ├── redis.py # Redis operations (caching, timestamps)
|
||||
│ ├── gates.py # Data quality gates and filtering
|
||||
│ └── mongodb.py # MongoDB operations (data loading)
|
||||
├── workflow/ # Temporal workflow definitions
|
||||
│ ├── scouter.py # Main data ingestion workflow
|
||||
│ ├── pi_web_api_scouter.py # PI Web API data ingestion workflow
|
||||
│ └── sub_workflows/ # Sub-workflow implementations
|
||||
│ └── core_scouter.py # Core data processing workflow
|
||||
├── worker/ # Worker implementation
|
||||
@@ -484,6 +591,7 @@ The Activities class combines multiple service classes through multiple inherita
|
||||
- **Redis**: Timestamp management, data caching, and temporary storage
|
||||
- **Gates**: Data quality validation and filtering logic
|
||||
- **MongoDB**: Data loading from raw collections
|
||||
- **API**: PI Web API tag value retrieval and data normalization
|
||||
|
||||
All activities support:
|
||||
- Comprehensive logging and error handling
|
||||
|
||||
Reference in New Issue
Block a user