diff --git a/README.md b/README.md index 0cad971..6e3a682 100644 --- a/README.md +++ b/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 diff --git a/scouter/activities/api.py b/scouter/activities/api.py index 045b415..eb2806a 100644 --- a/scouter/activities/api.py +++ b/scouter/activities/api.py @@ -63,6 +63,10 @@ class API(SientiaMonitoring): def close(self) -> None: """ Close the PI Web API client and shutdown monitoring services. + + This method performs cleanup operations: + - Closes the PI Web API client connection + - Shuts down SientiaMonitoring services (metrics, notifications) """ self.pi_web_api_client.close() SientiaMonitoring.shutdown(self) @@ -76,6 +80,11 @@ class API(SientiaMonitoring): for a set of configured tags. It returns the data as a list of dictionaries suitable for further processing in the workflow. + The timestamps are normalized to ensure consistency across all records in the + response. After converting timestamps to string format, all timestamps are + set to the maximum timestamp value (lexicographically) found in the dataset. + This ensures all records in a single batch share the same timestamp value. + Args: input_data (dict[str, Any]): Activity input parameters. Required fields: @@ -88,7 +97,7 @@ class API(SientiaMonitoring): Returns: list[dict]: List of data records, each containing: - - timestamp: Data point timestamp + - timestamp: Normalized timestamp string (all records share the same value) - name: Tag name - value: Numeric value - tag: WebId diff --git a/scouter/workflow/pi_web_api_scouter.py b/scouter/workflow/pi_web_api_scouter.py index 42cbce6..acaa1dc 100644 --- a/scouter/workflow/pi_web_api_scouter.py +++ b/scouter/workflow/pi_web_api_scouter.py @@ -33,26 +33,31 @@ class PIWebAPIScouter: This method orchestrates the complete data ingestion process from PI Web API: 1. Retrieves tag values from PI Web API using configured WebIds - 2. Validates and normalizes the retrieved data + 2. Validates and normalizes the retrieved data (timestamps are normalized) 3. Delegates data processing to the CoreScouter workflow + If no data is retrieved from the PI Web API, the workflow exits early without + invoking the CoreScouter workflow. + Args: input_data (dict[str, Any]): Configuration and parameters for the workflow execution. Required fields: + - schedule_name (str): Unique identifier for the data collection schedule - model_name (str): Name of the data model being processed - model_id (str): Unique identifier for the data model - - schedule_name (str): Unique identifier for the data collection schedule - - endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded') - - web_ids (dict[str, str | None]): Mapping of tag names to WebIds - - period (dict[str, str]): Time period configuration with 'start_time' - - api_timeout (int): Request timeout in seconds for PI Web API calls - - max_count (int, optional): Maximum data points per tag. Defaults to 1 + - pi_web_api_query (dict[str, Any]): PI Web API query configuration containing: + - endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded') + - period (str): Time period configuration (e.g., '*-1d', '*-1h') + - api_timeout (int): Request timeout in seconds for PI Web API calls + - max_count (int, optional): Maximum data points per tag. Defaults to 1 - trigger_laborious (bool): Flag to enable intensive data processing - filters (dict[str, str]): Data quality filters configuration - schema (str): Target database schema for data export - table_name (str): Target table name for data export - retention_time (int): Data retention period in Redis (seconds) - - model_tags (dict[str, Any]): Tag-specific configuration including: + - model_tags (dict[str, Any]): Tag-specific configuration mapping tag names + to WebIds and processing rules, including: + - webid (str): PI Web API WebId for the tag - data_range: [min, max] values for data validation - aggr_function: Aggregation method (avg, mdn, max, min, lts) - frequency: Data collection frequency in milliseconds diff --git a/scouter/workflow/scouter.py b/scouter/workflow/scouter.py index 52393e2..3ca3bbe 100644 --- a/scouter/workflow/scouter.py +++ b/scouter/workflow/scouter.py @@ -32,10 +32,14 @@ class Scouter: This method orchestrates the complete data ingestion process: 1. Retrieves the last processed timestamp from Redis - 2. Loads new data from MongoDB since the last timestamp - 3. Updates the last processed timestamp + 2. Loads new data from MongoDB since the last timestamp using collection name + format: `raw_{schedule_name}` + 3. Updates the last processed timestamp with the most recent data point 4. Delegates data processing to the CoreScouter workflow + If no new data is found in MongoDB, the workflow exits early without updating + the timestamp or invoking the CoreScouter workflow. + Args: input_data (dict[str, Any]): Configuration and parameters for the workflow execution. Required fields: diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index e395979..3aac96b 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -23,7 +23,8 @@ class CoreScouter: - Metrics collection and monitoring The workflow is designed for high-throughput data processing with configurable - quality gates and aggregation strategies. + quality gates and aggregation strategies. It is typically invoked as a child + workflow by parent workflows such as Scouter or PIWebAPIScouter. """ @workflow.run @@ -35,9 +36,14 @@ class CoreScouter: 1. Data Quality Gate: Applies configurable filters for data validation 2. Data Aggregation: Groups and aggregates data using specified functions 3. Data Grouping: Organizes data by tags and applies retention policies - 4. Data Export: Persists processed data to PostgreSQL + 4. Data Export: Persists processed data to PostgreSQL with timestamp conversion 5. Metrics Collection: Records processing metrics for monitoring + The workflow implements early exit conditions: + - If held_data is empty after grouping, the workflow exits without exporting + - If data export results in zero or negative affected_rows, the workflow exits + without writing metrics or storing debug packages + Args: input_data (dict[str, Any]): Complete workflow configuration and data. Required fields: @@ -53,6 +59,9 @@ class CoreScouter: - table_name (str): Target database table - retention_time (int): Redis data retention period (seconds) - model_tags (dict[str, Any]): Tag-specific processing rules + - fill_missing_tags (bool): Enable filling of missing tag values + - debug_data_package (bool, optional): Store data packages for debugging. + When True, stores both raw and processed data in MongoDB for debugging Returns: None: This workflow processes data but doesn't return results