diff --git a/.env.example b/.env.example index a4ba811..9992938 100644 --- a/.env.example +++ b/.env.example @@ -19,6 +19,10 @@ REDIS_PASSWORD="pass" TEMPORAL_HOST=localhost:7233 TEMPORAL_NAMESPACE=scouter +PI_WEB_API_BASE_URL="https://piwebapi.link.com/piwebapi" +PI_WEB_API_AUTH_TYPE="basic" +PI_WEB_API_AUTH_TOKEN="password" + LOG_LEVEL=INFO PROJECT_NAME=scouter diff --git a/requirements.txt b/requirements.txt index f9857c6..541df15 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,6 @@ psycopg2-binary sqlalchemy redis pymongo -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1 -prometheus-client \ No newline at end of file +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.7.1 +prometheus-client +pycurl \ No newline at end of file diff --git a/scouter/activities/activities.py b/scouter/activities/activities.py index 2f3f8f9..0cc3f50 100644 --- a/scouter/activities/activities.py +++ b/scouter/activities/activities.py @@ -9,12 +9,13 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.logger import Logger from sientia_do.temporal.activities.postgres import Postgres + from scouter.activities.api import API from scouter.activities.gates import Gates from scouter.activities.mongodb import MongoDB from scouter.activities.redis import Redis -class Activities(Postgres, Redis, Gates, MongoDB): +class Activities(Postgres, Redis, Gates, MongoDB, API): """ Unified activities class that combines multiple data processing services. @@ -24,6 +25,7 @@ class Activities(Postgres, Redis, Gates, MongoDB): - Redis operations for caching and temporary storage - Data quality gates and filtering - MongoDB operations for data retrieval + - PI Web API operations for external data ingestion - Notification handling and logging The class implements the multiple inheritance pattern to provide a unified @@ -35,6 +37,7 @@ class Activities(Postgres, Redis, Gates, MongoDB): postgres_config: dict[str, Any], redis_config: dict[str, Any], mongodb_config: dict[str, Any], + api_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler, ): @@ -48,6 +51,8 @@ class Activities(Postgres, Redis, Gates, MongoDB): Required fields: host, port, username, password mongodb_config (dict[str, Any]): MongoDB connection configuration. Required fields: connection_string, database_name + api_config (dict[str, Any]): PI Web API configuration. + Required fields: base_url, auth_type, auth_token logger (Logger): Logger instance for application logging notification_handler (NotificationHandler): Handler for system notifications """ @@ -100,6 +105,17 @@ class Activities(Postgres, Redis, Gates, MongoDB): metrics_controller=metrics_controller, ) + # Initialize API + API.__init__( + self, + base_url=api_config['base_url'], + auth_type=api_config['auth_type'], + auth_token=api_config['auth_token'], + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + self.pod_id = getenv('HOSTNAME', 'localhost') def shutdown(self): @@ -113,3 +129,4 @@ class Activities(Postgres, Redis, Gates, MongoDB): MongoDB.close(self) Redis.close(self) Gates.close(self) + API.close(self) diff --git a/scouter/activities/api.py b/scouter/activities/api.py new file mode 100644 index 0000000..e70f19b --- /dev/null +++ b/scouter/activities/api.py @@ -0,0 +1,139 @@ +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + import traceback + from typing import Any + + from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.notifications.models import NotificationLevel + from sientia_do.observability.logger import Logger + from sientia_do.observability.metrics_controller import MetricsController + from sientia_do.observability.sientia_monitoring import SientiaMonitoring + from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ + + from scouter.utils.clients.pi_web_api_client import PIWebAPIClient + + +class API(SientiaMonitoring): + """ + PI Web API operations for data retrieval. + + This class provides Temporal activities for interacting with the PI Web API + to retrieve tag values and historical data. It implements: + - Tag value retrieval from PI Web API endpoints + - Data quality filtering and validation + - Error handling with notifications + - Metrics collection for monitoring + + The class wraps the PIWebAPIClient to provide Temporal-aware activity methods + that can be used in workflow orchestration. + """ + + def __init__( + self, + base_url: str, + auth_type: str, + auth_token: str, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + ) -> None: + """ + Initialize API activity with PI Web API client. + + Args: + base_url (str): Base URL of the PI Web API server + auth_type (str): Authentication type ('basic' or 'bearer') + auth_token (str): Authentication token + logger (Logger): Logger instance for operation logging + notification_handler (NotificationHandler): Handler for system notifications + metrics_controller (MetricsController): Controller for metrics collection + """ + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) + self.pi_web_api_client = PIWebAPIClient( + base_url=base_url, + auth_config={ + 'type': auth_type, + 'token': auth_token, + }, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + def close(self) -> None: + """ + Close the PI Web API client and shutdown monitoring services. + """ + self.pi_web_api_client.close() + SientiaMonitoring.shutdown(self) + + @activity.defn(name='get_tag_values') + async def get_tag_values(self, input_data: dict[str, Any]) -> list[dict]: + """ + Retrieve tag values from PI Web API for specified WebIds. + + This activity fetches historical or real-time data from the PI Web API + for a set of configured tags. It returns the data as a list of dictionaries + suitable for further processing in the workflow. + + Args: + input_data (dict[str, Any]): Activity input parameters. + Required fields: + - metadata (dict[str, Any]): Workflow execution metadata + - endpoint (str): PI Web API endpoint path + - web_ids (dict[str, str | None]): Tag names mapped to WebIds + - period (dict[str, str]): Time period with 'start_time' field + - api_timeout (int): Request timeout in seconds + - max_count (int, optional): Maximum data points per tag. Defaults to 1 + + Returns: + list[dict]: List of data records, each containing: + - timestamp: Data point timestamp + - name: Tag name + - value: Numeric value + - tag: WebId + + Raises: + PIMSRequestError: If API request fails + Exception: If data retrieval or processing fails + """ + metadata = input_data['metadata'] + endpoint = input_data['endpoint'] + web_ids = input_data['web_ids'] + period = input_data['period'] + max_count = input_data.get('max_count', 1) + api_timeout = input_data['api_timeout'] + + self.info(f'Getting tag values from {endpoint}', metadata=metadata) + self.debug(f'Web IDs: {web_ids}', metadata=metadata) + try: + latest_values = await self.pi_web_api_client.get_latest_values_df( + endpoint=endpoint, + web_ids=web_ids, + start_time=period, + max_count=max_count, + metadata=metadata, + timeout=api_timeout, + ) + + except Exception as e: + await self.send_notification_async( + metadata=metadata, + notification_id='PI_WEB_API_REQUEST_ERROR', + message=f'Error getting tag values from PI Web API: {e}', + block='get_tag_values', + level=NotificationLevel.ERROR, + attachment_content=traceback.format_exc(), + ) + raise e + + latest_values['timestamp'] = latest_values['timestamp'].dt.strftime(DATETIME_FORMAT_WITH_TZ) + + # Normalize the package timestamp + latest_values['timestamp'] = latest_values['timestamp'].max() + + self.debug(f'Latest values: {latest_values}', metadata=metadata) + self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata) + + return latest_values.to_dict(orient='records') diff --git a/scouter/metrics.py b/scouter/metrics.py index b25fa46..3ef2288 100644 --- a/scouter/metrics.py +++ b/scouter/metrics.py @@ -1,4 +1,5 @@ -from prometheus_client import Counter, Gauge +from prometheus_client import Counter, Gauge, Histogram +from sientia_do.observability.metrics import CORE_LABELS as SIENTIA_CORE_LABELS # Application health and status metrics APP_UP = Gauge( @@ -23,3 +24,23 @@ TAG_CHANGES_MONITOR = Gauge( 'Current value change of each tag', [*CORE_LABELS, 'tag_name'], ) + + +# Generic REST client metrics +GENERIC_REST_CLIENT_LAG = Histogram( + 'scouter_generic_rest_client_lag', + 'Lag time for a request to a generic REST client', + SIENTIA_CORE_LABELS, +) + +GENERIC_REST_READ_COUNT = Counter( + 'scouter_generic_rest_client_read_count', + 'Number of reads from a generic REST client', + SIENTIA_CORE_LABELS, +) + +GENERIC_REST_READ_ERROR_COUNT = Counter( + 'scouter_generic_rest_client_read_error_count', + 'Number of read errors from a generic REST client', + SIENTIA_CORE_LABELS, +) diff --git a/scouter/utils/clients/pi_web_api_client.py b/scouter/utils/clients/pi_web_api_client.py new file mode 100644 index 0000000..d18528a --- /dev/null +++ b/scouter/utils/clients/pi_web_api_client.py @@ -0,0 +1,335 @@ +import io +import json +import time +import warnings +from typing import Any +from urllib.parse import urlencode + +import pandas as pd +import pycurl +from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring + +from scouter import metrics + +warnings.simplefilter('ignore') # Ignore warnings such as 'verify=False' + + +class PIMSRequestError(Exception): + """Generic error for failed requests to PI Web API using pycurl.""" + + pass + + +class PIWebAPIClient(SientiaMonitoring): + """ + Client for interacting with the PI Web API. + + This class provides a robust interface for querying historical and real-time + data from OSIsoft PI systems through the PI Web API. It implements: + - Asynchronous HTTP requests using pycurl + - Authentication support (Basic and Bearer) + - Automatic data normalization and timestamp handling + - Comprehensive error handling and monitoring + - Metrics collection for observability + + The client is designed for high-performance data retrieval with proper + connection management and error recovery mechanisms. + """ + + def __init__( + self, + base_url: str, + auth_config: dict[str, Any], + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + headers_config: dict[str, Any] | None = None, + ) -> None: + """ + Initialize the PI Web API client with connection parameters. + + Args: + base_url (str): Base URL of the PI Web API server + auth_config (dict[str, Any]): Authentication configuration. + Required fields: + - type (str): Authentication type ('basic' or 'bearer') + - token (str): Authentication token + logger (Logger): Logger instance for operation logging + notification_handler (NotificationHandler): Handler for system notifications + metrics_controller (MetricsController): Controller for metrics collection + headers_config (dict[str, Any], optional): HTTP headers configuration. + Default headers include content-type, accept, and x-requested-with + max_concurrency (int, optional): Maximum number of concurrent requests. Defaults to 8 + """ + if headers_config is None: + headers_config = { + 'Content-Type': 'application/json', + 'Accept': 'application/json', + 'x-requested-with': 'XMLHttpRequest', + } + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) + + self.base_url = base_url.rstrip('/') + + # auth_config spec: + # 'type': 'basic' or 'bearer', + # 'token': 'token', + self.auth_config = auth_config + self.auth_config['type'] = self.auth_config['type'].lower() + self.headers: dict[str, str] = headers_config + + self.authenticate() + + def close(self) -> None: + """ + Close the client and shutdown monitoring services. + """ + SientiaMonitoring.shutdown(self) + + def _to_clean_timestamp(self, series: pd.Series) -> pd.Series: + """ + Convert a Series of timestamps to datetime, UTC, and round to the nearest second. + + Args: + series (pd.Series): Series containing timestamp values + + Returns: + pd.Series: Cleaned timestamp series in UTC, floored to seconds + """ + series = pd.to_datetime(series, utc=True, errors='coerce') + return series.dt.floor('s') + + def _extract_numeric(self, value: Any) -> float | None: + """ + Normalize a value (potentially nested) to float. + + This method handles PI Web API response values that may be nested + in dictionaries or other structures, extracting the numeric value. + + Args: + value (Any): Value to extract and normalize + + Returns: + float | None: Numeric value as float, or None if conversion fails + """ + if isinstance(value, dict): + value = value.get('Value', value) + return pd.to_numeric(value, errors='coerce') + + def authenticate(self): + """ + Configure authentication headers based on auth_config. + + This method sets up the Authorization header using either Basic or Bearer + authentication based on the configured authentication type. + + Raises: + ValueError: If authentication type is not 'basic' or 'bearer' + """ + self.logger.info(f'Authenticating with {self.auth_config["type"]} authentication') + + if self.auth_config['type'] == 'basic': + self.headers['Authorization'] = f'Basic {self.auth_config["token"]}' + elif self.auth_config['type'] == 'bearer': + self.headers['Authorization'] = f'Bearer {self.auth_config["token"]}' + else: + raise ValueError(f'Invalid authentication type: {self.auth_config["type"]}') + + async def _curl_get_json( + self, + url: str, + params: list[tuple[str, str]] | None = None, + timeout: int = 30, + verify: bool = True, + metadata: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """ + Perform a GET request using pycurl and return the decoded JSON response. + + This method executes an asynchronous HTTP GET request with proper error handling, + metrics collection, and timeout management. It automatically tracks request + latency and emits monitoring metrics. + + Args: + url (str): Target URL for the GET request + params (list[tuple[str, str]], optional): Query parameters as list of tuples. + Each tuple contains (parameter_name, parameter_value) + timeout (int, optional): Request timeout in seconds. Defaults to 30 + verify (bool, optional): Verify SSL certificates. Defaults to True + metadata (dict[str, Any], optional): Workflow execution metadata for tracking + + Returns: + dict[str, Any]: Parsed JSON response body + + Raises: + PIMSRequestError: If HTTP error, connection error, or JSON parsing error occurs + """ + if metadata is None: + metadata = {} + + buffer = io.BytesIO() + c = pycurl.Curl() + + core_labels = self.get_core_labels(metadata=metadata, operation_type='get_json') + + try: + if params: + query_string = urlencode(params, doseq=True) + full_url = f'{url}?{query_string}' + else: + full_url = url + + c.setopt(pycurl.URL, full_url.encode('utf-8')) + c.setopt(pycurl.WRITEDATA, buffer) + + # Configure HTTP headers + header_list = [f'{k}: {v}' for k, v in self.headers.items()] + if header_list: + c.setopt(pycurl.HTTPHEADER, header_list) + + # Set request timeout + c.setopt(pycurl.TIMEOUT, timeout) + + # Configure SSL verification + if not verify: + c.setopt(pycurl.SSL_VERIFYPEER, 0) + c.setopt(pycurl.SSL_VERIFYHOST, 0) + + start_time = time.time() + try: + c.perform() + except Exception as e: + await self.emit_metric( + metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT, + tags=core_labels, + ) + raise e + await self.observe_lag( + start_time=start_time, + metric_object=metrics.GENERIC_REST_CLIENT_LAG, + tags=core_labels, + ) + + status_code = c.getinfo(pycurl.RESPONSE_CODE) + body = buffer.getvalue().decode('utf-8', errors='replace') + + if status_code >= 400: + await self.emit_metric( + metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT, + tags=core_labels, + ) + raise PIMSRequestError(f"HTTP {status_code} calling '{full_url}': {body[:200]}") + + await self.emit_metric( + metric_object=metrics.GENERIC_REST_READ_COUNT, + tags=core_labels, + ) + try: + return json.loads(body) + except json.JSONDecodeError as e: + raise PIMSRequestError( + f"Error decoding JSON response from '{full_url}': {e}; body: {body[:200]}" + ) from e + + except pycurl.error as e: + raise PIMSRequestError(f"Connection error calling '{url}': {e}") from e + finally: + c.close() + + async def get_latest_values_df( + self, + web_ids: dict[str, dict[str, str]], + endpoint: str, + timeout: int = 30, + start_time: str = '*-1d', + end_time: str = '*', + max_count: int | None = 1, + metadata: dict[str, Any] | None = None, + ) -> pd.DataFrame: + """ + Retrieve historical values for multiple WebIds using PI Web API streamsets. + + This method queries the PI Web API's /streamsets/recorded endpoint to fetch + historical data for multiple tags simultaneously. It returns a normalized + DataFrame with timestamps, tag names, values, and WebIds. + + Args: + web_ids (dict[str, str | None]): Dictionary mapping tag names to their WebIds. + None values are filtered out before querying + endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded') + timeout (int, optional): Request timeout in seconds. Defaults to 30 + start_time (str, optional): Start time in PI Web API format (e.g., "*-50d"). + Defaults to "*-1d" (1 day ago) + end_time (str, optional): End time in PI Web API format (e.g., "*"). + Defaults to "*" (current time) + max_count (int, optional): Maximum number of data points per series. + Defaults to 1. If None, maxCount parameter is not sent + metadata (dict[str, Any], optional): Workflow execution metadata for tracking + + Returns: + pd.DataFrame: DataFrame with columns: + - timestamp: Cleaned timestamp (UTC, floored to seconds) + - name: Tag name + - value: Numeric value (normalized) + - tag: WebId of the tag + Returns empty DataFrame if no data is found + + Raises: + PIMSRequestError: If API request fails or returns invalid data + """ + if metadata is None: + metadata = {} + + url = f'{self.base_url}{endpoint}' + + # Build query parameters with WebIds (filtering out None values) + params: list[tuple[str, str]] = [('webid', web_id['webid']) for web_id in web_ids.values()] + params.extend( + [ + ('startTime', start_time), + ('endtime', end_time), + ('selectedFields', 'Items.Name;Items.Items.Timestamp;Items.Items.Value'), + ] + ) + + params.append(('maxCount', str(max_count))) + + data = await self._curl_get_json( + url=url, + params=params, + timeout=timeout, + verify=False, + metadata=metadata, + ) + + raw_data = data.get('Items', []) + + records = [] + for entry in raw_data: + tag_name = entry.get('Name') + series_items = entry.get('Items', []) + for it in series_items: + if isinstance(it, dict) and 'Timestamp' in it and 'Value' in it: + ts = it.get('Timestamp') + val = it.get('Value') + web_id = web_ids[tag_name]['webid'] + if ts is not None: + records.append( + { + 'timestamp': ts, + 'name': tag_name, + 'value': self._extract_numeric(val), + 'tag': web_id, + } + ) + + if not records: + return pd.DataFrame() + + df = pd.DataFrame.from_records(records) + df['timestamp'] = self._to_clean_timestamp(df['timestamp']) + + return df diff --git a/scouter/utils/connectors_config.py b/scouter/utils/connectors_config.py index 4264aa1..da712d9 100644 --- a/scouter/utils/connectors_config.py +++ b/scouter/utils/connectors_config.py @@ -83,6 +83,23 @@ def build_mongodb_config() -> dict[str, Any]: } +def build_api_config() -> dict[str, Any]: + """ + Build API connection configuration from environment variables. + + Returns: + dict[str, Any]: API configuration dictionary with keys: + - base_url: API base URL (default: https://pi.example.com) + - auth_type: API authentication type (default: basic) + - auth_token: API authentication token (default: None) + """ + return { + 'base_url': getenv('PI_WEB_API_BASE_URL', 'https://pi.example.com'), + 'auth_type': getenv('PI_WEB_API_AUTH_TYPE', 'basic'), + 'auth_token': getenv('PI_WEB_API_AUTH_TOKEN', None), + } + + def build_druid_config() -> dict[str, Any]: """ Build Apache Druid connection configuration from environment variables. diff --git a/scouter/worker/prepare_worker.py b/scouter/worker/prepare_worker.py index 09fced1..c8b68f2 100644 --- a/scouter/worker/prepare_worker.py +++ b/scouter/worker/prepare_worker.py @@ -1,7 +1,9 @@ import os +import re from collections.abc import Sequence from typing import Any +from sientia_do.observability.logger import Logger from temporalio.client import Client from temporalio.worker import PollerBehaviorAutoscaling, Worker @@ -19,14 +21,24 @@ parameters = [ ] +def camel_to_snake(text: str) -> str: + """Convert camelCase or PascalCase to snake_case.""" + text = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', text) + text = re.sub('([a-z0-9])([A-Z])', r'\1_\2', text) + return text.lower() + + def prepare_worker( main_workflow: type, other_workflows: Sequence[type], activities: Sequence[Any], temporal_client: Client, + logger: Logger, ) -> Worker: main_workflow_name = main_workflow.__name__.upper() + queue_name = f'{camel_to_snake(main_workflow.__name__)}-queue' + local_workflow_parameters = {} for parameter in parameters: @@ -34,9 +46,11 @@ def prepare_worker( os.getenv(main_workflow_name + '_' + parameter[0], parameter[1]) ) + logger.info(f'Preparing worker for {main_workflow_name} with queue {queue_name}') + return Worker( temporal_client, - task_queue='scouter-queue', + task_queue=queue_name, workflows=[main_workflow, *other_workflows], activities=[*activities], max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'], diff --git a/scouter/worker/worker.py b/scouter/worker/worker.py index cd333f4..4de701a 100644 --- a/scouter/worker/worker.py +++ b/scouter/worker/worker.py @@ -15,10 +15,12 @@ with workflow.unsafe.imports_passed_through(): from scouter import metrics from scouter.activities.activities import Activities from scouter.utils.connectors_config import ( + build_api_config, build_mongodb_config, build_postgres_config, build_redis_config, ) + from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter from scouter.workflow.scouter import Scouter from scouter.workflow.sub_workflows.core_scouter import CoreScouter @@ -103,6 +105,7 @@ async def main(): postgres_config=build_postgres_config(), redis_config=build_redis_config(), mongodb_config=build_mongodb_config(), + api_config=build_api_config(), ) logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) @@ -137,7 +140,23 @@ async def main(): activities.write_metrics, activities.store_data_package, ], - ) + logger=logger, + ), + prepare_worker( + temporal_client=temporal_client, + main_workflow=PIWebAPIScouter, + other_workflows=[CoreScouter], + activities=[ + activities.get_tag_values, + activities.data_quality_gate, + activities.aggregate_data, + activities.group_and_hold_data, + activities.export_data_to_postgres, + activities.write_metrics, + activities.store_data_package, + ], + logger=logger, + ), ] handlers = [] diff --git a/scouter/workflow/pi_web_api_scouter.py b/scouter/workflow/pi_web_api_scouter.py new file mode 100644 index 0000000..6582e4e --- /dev/null +++ b/scouter/workflow/pi_web_api_scouter.py @@ -0,0 +1,103 @@ +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from datetime import timedelta + from typing import Any + + from sientia_do.temporal.policies import retry_policy + + from scouter.activities.activities import Activities + + +@workflow.defn(name='pi_web_api_scouter') +class PIWebAPIScouter: + """ + PI Web API Scouter workflow that orchestrates data ingestion from PI systems. + + This workflow serves as the entry point for PI Web API data processing pipelines. + Unlike the standard Scouter that loads from MongoDB, this workflow directly queries + PI Web API endpoints to retrieve tag values and processes them for downstream use. + + The workflow implements a direct API ingestion pattern with: + - Real-time data retrieval from PI Web API + - Configurable time periods and data point limits + - Error handling and retry policies + - Child workflow orchestration for data processing + - Integration with CoreScouter for standardized processing + """ + + @workflow.run + async def run(self, input_data: dict[str, Any]) -> None: + """ + Execute the PI Web API Scouter workflow. + + 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 + 3. Delegates data processing to the CoreScouter workflow + + Args: + input_data (dict[str, Any]): Configuration and parameters for the workflow execution. + Required fields: + - 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 + - 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: + - data_range: [min, max] values for data validation + - aggr_function: Aggregation method (avg, mdn, max, min, lts) + - frequency: Data collection frequency in milliseconds + - topics: List of Kafka topics for data routing + + Returns: + None: This workflow doesn't return data, it orchestrates data processing + + Raises: + WorkflowExecutionError: If workflow execution fails + ActivityExecutionError: If any activity fails after retry attempts + PIMSRequestError: If PI Web API request fails + """ + + input_data['workflow_name'] = 'scouter' + + metadata = { + 'metadata': { + 'model_id': input_data['model_id'], + 'model_name': input_data['model_name'], + 'schedule_name': input_data['schedule_name'], + 'workflow_name': input_data['workflow_name'], + } + } + + pi_web_api_query = input_data['pi_web_api_query'] + + data = await workflow.execute_local_activity_method( + Activities.get_tag_values, + { + **metadata, + 'endpoint': pi_web_api_query['endpoint'], + 'web_ids': input_data['model_tags'], + 'period': pi_web_api_query['period'], + 'max_count': pi_web_api_query.get('max_count', 1), + 'api_timeout': pi_web_api_query['api_timeout'], + }, + start_to_close_timeout=timedelta(seconds=60), + retry_policy=retry_policy, + ) + + if not data: + return + + input_data['data'] = data + input_data['metadata'] = metadata + + await workflow.execute_child_workflow('subworkflow.core_scouter', input_data) diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 1bbc4cb..2e9b1c1 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -102,7 +102,7 @@ class CoreScouter: if held_data == {}: return - await workflow.execute_activity_method( + data_exported = await workflow.execute_activity_method( Activities.export_data_to_postgres, { **metadata, @@ -110,11 +110,16 @@ class CoreScouter: 'table_name': input_data['table_name'], 'data': held_data, 'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ}, + 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60), ) + if data_exported.get('affected_rows', 0) <= 0: + return + await workflow.execute_activity_method( Activities.write_metrics, { diff --git a/tests.ipynb b/tests.ipynb new file mode 100644 index 0000000..79f937d --- /dev/null +++ b/tests.ipynb @@ -0,0 +1,984 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 12, + "id": "9d16b24a", + "metadata": {}, + "outputs": [], + "source": [ + "from datetime import timedelta\n", + "from typing import Any\n", + "from temporalio import client\n", + "from temporalio.client import WorkflowHandle\n", + "\n", + "\n", + "async def start_workflow_advanced(\n", + " temporal_client: client.Client,\n", + " workflow_name: str,\n", + " workflow_input: dict[str, Any],\n", + " workflow_id: str,\n", + " task_queue: str,\n", + " execution_timeout: timedelta | None = None,\n", + " run_timeout: timedelta | None = None,\n", + " task_timeout: timedelta | None = None,\n", + ") -> WorkflowHandle:\n", + " handle = await temporal_client.start_workflow(\n", + " workflow=workflow_name,\n", + " arg=workflow_input,\n", + " id=workflow_id or f\"{workflow_name}-{id(workflow_input)}\",\n", + " task_queue=task_queue,\n", + " execution_timeout=execution_timeout,\n", + " run_timeout=run_timeout,\n", + " task_timeout=task_timeout,\n", + " )\n", + " \n", + " return handle" + ] + }, + { + "cell_type": "code", + "execution_count": 13, + "id": "5e344fb0", + "metadata": {}, + "outputs": [], + "source": [ + "input_data = {\n", + " \"debug_data_package\": False,\n", + " \"execution_timeout_seconds\": 300,\n", + " \"fill_missing_tags\": False,\n", + " \"filters\": {\n", + " \"NULL_VALUES_FILTER\": {\n", + " \"policy\": \"DISCARD\"\n", + " },\n", + " \"OUT_OF_BOUNDS_FILTER\": {\n", + " \"policy\": \"DISCARD\"\n", + " }\n", + " },\n", + " \"frequency\": \"30s\",\n", + " \"max_retry_policy\": 1,\n", + " \"model_config\": {\n", + " \"predict_flavor\": \"sklearn\",\n", + " \"retention_minutes\": 0,\n", + " \"target\": \"CI-W3A05F1\",\n", + " \"transform_flavor\": \"sklearn\"\n", + " },\n", + " \"model_id\": \"10\",\n", + " \"model_name\": \"Pi Web API Test Model\",\n", + " \"model_tags\": {\n", + " \"CI-W3W03S1\": {\n", + " \"aggr_func\": \"avg\",\n", + " \"data_range\": [\n", + " -100000,\n", + " 100000\n", + " ],\n", + " \"webid\": \"F1DP-7fYgsRTtUOa7V9NIwSujATFUAAAUElIQVZDXENJLVczVzAzUzE\"\n", + " },\n", + " \"CI-W3A05F1\": {\n", + " \"aggr_func\": \"lts\",\n", + " \"data_range\": [\n", + " -100000,\n", + " 100000\n", + " ],\n", + " \"webid\": \"F1DP-7fYgsRTtUOa7V9NIwSujAkVMAAAUElIQVZDXENJLVczQTA1RjE\"\n", + " }\n", + " },\n", + " \"pi_web_api_query\": {\n", + " \"endpoint\": \"/streamsets/recorded\",\n", + " \"period\": \"*-1d\",\n", + " \"max_count\": 1,\n", + " \"api_timeout\": 5\n", + " },\n", + " \"offset\": \"0m\",\n", + " \"retention_time\": 3600,\n", + " \"schedule_name\": \"pi-web-api-scouter-test\",\n", + " \"schema\": \"sientia_data\",\n", + " \"table_name\": \"laborious_data\",\n", + " \"task_timeout_seconds\": 300,\n", + " \"trigger_laborious\": False,\n", + " \"updated_at\": \"2025-08-13 18:35:01.600000+0000\",\n", + " \"workflow_type\": \"pi_web_api_scouter\"\n", + "}" + ] + }, + { + "cell_type": "code", + "execution_count": 14, + "id": "9350bff3", + "metadata": {}, + "outputs": [], + "source": [ + "from temporalio import client\n", + "\n", + "temporal_client = await client.Client.connect(\n", + " target_host=\"localhost:7233\",\n", + " namespace=\"scouter\"\n", + ")\n" + ] + }, + { + "cell_type": "code", + "execution_count": 15, + "id": "45712d7a", + "metadata": {}, + "outputs": [], + "source": [ + "\n", + "import datetime\n", + "\n", + "now = datetime.datetime.now()\n", + "\n", + "handle = await start_workflow_advanced(\n", + " temporal_client=temporal_client,\n", + " workflow_name='pi_web_api_scouter',\n", + " workflow_input=input_data,\n", + " workflow_id='test_workflow_id_' + now.strftime('%Y%m%d%H%M%S'),\n", + " task_queue='pi-web-api-scouter-queue',\n", + " execution_timeout=timedelta(seconds=30),\n", + " run_timeout=timedelta(seconds=30),\n", + " task_timeout=timedelta(seconds=30),\n", + ")\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": 16, + "id": "d065d0de", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "pi-web-api-scouter\n" + ] + } + ], + "source": [ + "import re\n", + "def camel_to_kebab(text: str) -> str:\n", + " \"\"\"Convert camelCase or PascalCase to kebab-case.\"\"\"\n", + " text = re.sub('(.)([A-Z][a-z]+)', r'\\1-\\2', text)\n", + " text = re.sub('([a-z0-9])([A-Z])', r'\\1-\\2', text)\n", + " return text.lower()\n", + "\n", + "print(camel_to_kebab('PiWebApiScouter'))\n" + ] + }, + { + "cell_type": "code", + "execution_count": 17, + "id": "72af4236", + "metadata": {}, + "outputs": [ + { + "ename": "KeyError", + "evalue": "'Items'", + "output_type": "error", + "traceback": [ + "\u001b[31m---------------------------------------------------------------------------\u001b[39m", + "\u001b[31mKeyError\u001b[39m Traceback (most recent call last)", + "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[17]\u001b[39m\u001b[32m, line 42\u001b[39m\n\u001b[32m 39\u001b[39m \u001b[38;5;28;01mfor\u001b[39;00m tag \u001b[38;5;129;01min\u001b[39;00m TAG_NAMES:\n\u001b[32m 40\u001b[39m response = requests.get(url.replace(\u001b[33m'\u001b[39m\u001b[38;5;132;01m{tag}\u001b[39;00m\u001b[33m'\u001b[39m, tag), headers=headers).json()\n\u001b[32m 41\u001b[39m web_ids[tag] = {\n\u001b[32m---> \u001b[39m\u001b[32m42\u001b[39m \u001b[33m'\u001b[39m\u001b[33mwebid\u001b[39m\u001b[33m'\u001b[39m: \u001b[43mresponse\u001b[49m\u001b[43m[\u001b[49m\u001b[33;43m'\u001b[39;49m\u001b[33;43mItems\u001b[39;49m\u001b[33;43m'\u001b[39;49m\u001b[43m]\u001b[49m[\u001b[32m0\u001b[39m][\u001b[33m'\u001b[39m\u001b[33mWebId\u001b[39m\u001b[33m'\u001b[39m],\n\u001b[32m 43\u001b[39m \u001b[33m'\u001b[39m\u001b[33maggr_func\u001b[39m\u001b[33m'\u001b[39m: \u001b[33m'\u001b[39m\u001b[33mlts\u001b[39m\u001b[33m'\u001b[39m,\n\u001b[32m 44\u001b[39m \u001b[33m'\u001b[39m\u001b[33mdata_range\u001b[39m\u001b[33m'\u001b[39m: [-\u001b[32m100000\u001b[39m, \u001b[32m100000\u001b[39m],\n\u001b[32m 45\u001b[39m }\n\u001b[32m 46\u001b[39m sleep(\u001b[32m0.5\u001b[39m)\n", + "\u001b[31mKeyError\u001b[39m: 'Items'" + ] + } + ], + "source": [ + "import requests\n", + "from time import sleep\n", + "\n", + "# Obter web id das seguintes tags:\n", + "TAG_NAMES = [\n", + " \"CI-W3A05F1\",\n", + " \"CI-W3W03S1\", \"CI-W3W03I1\", \"CI-W3K01T1\", \"CI-W3W01A3\",\n", + " \"CI-W3W01A2\", \"CI-W3W01A1\", \"CI-J3P01T1A\", \"CI-W3A50T1\", \"CI-W3A55T1\",\n", + " \"CI-W3A55P1\", \"CI-W3V33P1\", \"CI-W3E01F1\", \"CI-W3A50A3\", \"CI-W3A50A2\",\n", + " \"CI-W3A50A1\", \"CI-W3A50P1\", \"CI-W3W01P1\", \"CI-W3A71P1\", \"CI-W3W01P2\",\n", + " \"CI-W3A71P2\", \"CI-W3A71P3\", \"CI-J3J01S1\", \"CI-W3P17S1\", \"CI-J3P03S1\",\n", + " \"CI-W3K01S1\", \"CI-W3K14P1\", \"CI-W3K01T4\", \"CI-W3K01T2\", \n", + "\n", + " \"CI-W3FARCI_FSC\",\n", + " \"CI-W3FARCI_MA\",\n", + " \"CI-W3FARCI_MS\",\n", + " \"CI-W3FARCI_P100\",\n", + " \"CI-W3FARCI_p170\",\n", + "\n", + " \"CI-W3CLK_C3S\",\n", + " \"CI-W3CLK_C3S_EXP\",\n", + " \"CI-W3CLK_C3S_MD_EXP\",\n", + " \"CI-W3CLK_C3S_MD_PETRO\",\n", + " \"CI-W3V04P3\", \"CI-W3V04P1\",\n", + " \"CI-W3W01G1\"\n", + "]\n", + "\n", + "url = 'https://pivision.votorantimcimentos.com/piwebapi/dataservers/F1DS-7fYgsRTtUOa7V9NIwSujAUElIQVZD/points?namefilter={tag}'\n", + "\n", + "headers = {\n", + " 'Content-Type': 'application/json',\n", + " 'Accept': 'application/json',\n", + " 'X-Requested-With': 'piwebapistreams', # Header recomendado pelo PI Web API\n", + " 'Authorization': \"\"\n", + "}\n", + "\n", + "web_ids = {}\n", + "\n", + "for tag in TAG_NAMES:\n", + " response = requests.get(url.replace('{tag}', tag), headers=headers).json()\n", + " web_ids[tag] = {\n", + " 'webid': response['Items'][0]['WebId'],\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000],\n", + " }\n", + " sleep(0.5)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "55793801", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "{'CI-W3A05F1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAkVMAAAUElIQVZDXENJLVczQTA1RjE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W03S1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujATFUAAAUElIQVZDXENJLVczVzAzUzE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W03I1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAS1UAAAUElIQVZDXENJLVczVzAzSTE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3K01T1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAklQAAAUElIQVZDXENJLVczSzAxVDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W01A3': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAOFUAAAUElIQVZDXENJLVczVzAxQTM',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W01A2': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAN1UAAAUElIQVZDXENJLVczVzAxQTI',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W01A1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujANlUAAAUElIQVZDXENJLVczVzAxQTE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-J3P01T1A': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAYUUAAAUElIQVZDXENJLUozUDAxVDFB',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A50T1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujArFMAAAUElIQVZDXENJLVczQTUwVDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A55T1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAxFMAAAUElIQVZDXENJLVczQTU1VDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A55P1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAwVMAAAUElIQVZDXENJLVczQTU1UDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3V33P1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAJVUAAAUElIQVZDXENJLVczVjMzUDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3E01F1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAzY0CAAUElIQVZDXENJLVczRTAxRjE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A50A3': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAqVMAAAUElIQVZDXENJLVczQTUwQTM',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A50A2': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAqFMAAAUElIQVZDXENJLVczQTUwQTI',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A50A1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAp1MAAAUElIQVZDXENJLVczQTUwQTE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A50P1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAq1MAAAUElIQVZDXENJLVczQTUwUDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W01P1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAPlUAAAUElIQVZDXENJLVczVzAxUDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A71P1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAy1MAAAUElIQVZDXENJLVczQTcxUDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W01P2': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAP1UAAAUElIQVZDXENJLVczVzAxUDI',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A71P2': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAzFMAAAUElIQVZDXENJLVczQTcxUDI',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3A71P3': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAzVMAAAUElIQVZDXENJLVczQTcxUDM',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-J3J01S1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAMUUAAAUElIQVZDXENJLUozSjAxUzE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3P17S1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujA7VQAAAUElIQVZDXENJLVczUDE3UzE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-J3P03S1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAa0UAAAUElIQVZDXENJLUozUDAzUzE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3K01S1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAkFQAAAUElIQVZDXENJLVczSzAxUzE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3K14P1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAuFQAAAUElIQVZDXENJLVczSzE0UDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3K01T4': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAl1QAAAUElIQVZDXENJLVczSzAxVDQ',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3K01T2': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAlVQAAAUElIQVZDXENJLVczSzAxVDI',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3FARCI_FSC': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAgVQAAAUElIQVZDXENJLVczRkFSQ0lfRlND',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3FARCI_MA': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAg1QAAAUElIQVZDXENJLVczRkFSQ0lfTUE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3FARCI_MS': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAhVQAAAUElIQVZDXENJLVczRkFSQ0lfTVM',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3FARCI_P100': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAiFQAAAUElIQVZDXENJLVczRkFSQ0lfUDEwMA',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3FARCI_p170': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAiVQAAAUElIQVZDXENJLVczRkFSQ0lfUDE3MA',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3CLK_C3S': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujA4lMAAAUElIQVZDXENJLVczQ0xLX0MzUw',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3CLK_C3S_EXP': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujA41MAAAUElIQVZDXENJLVczQ0xLX0MzU19FWFA',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3CLK_C3S_MD_EXP': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujA5VMAAAUElIQVZDXENJLVczQ0xLX0MzU19NRF9FWFA',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3CLK_C3S_MD_PETRO': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujA5lMAAAUElIQVZDXENJLVczQ0xLX0MzU19NRF9QRVRSTw',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3V04P3': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAElUAAAUElIQVZDXENJLVczVjA0UDM',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3V04P1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAEFUAAAUElIQVZDXENJLVczVjA0UDE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]},\n", + " 'CI-W3W01G1': {'webid': 'F1DP-7fYgsRTtUOa7V9NIwSujAOlUAAAUElIQVZDXENJLVczVzAxRzE',\n", + " 'aggr_func': 'lts',\n", + " 'data_range': [-100000, 100000]}}" + ] + }, + "execution_count": 23, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "web_ids" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "2e0e3d5a", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "{'Links': {},\n", + " 'Items': [{'WebId': 'F1DP-7fYgsRTtUOa7V9NIwSujAkVMAAAUElIQVZDXENJLVczQTA1RjE',\n", + " 'Name': 'CI-W3A05F1',\n", + " 'Path': '\\\\\\\\PIHAVC\\\\CI-W3A05F1',\n", + " 'Links': {'Source': 'https://pivision.votorantimcimentos.com/piwebapi/points/F1DP-7fYgsRTtUOa7V9NIwSujAkVMAAAUElIQVZDXENJLVczQTA1RjE'},\n", + " 'Items': [{'Timestamp': '2025-12-17T18:12:49.2170104Z',\n", + " 'Value': 273.3339,\n", + " 'UnitsAbbreviation': '',\n", + " 'Good': True,\n", + " 'Questionable': False,\n", + " 'Substituted': False,\n", + " 'Annotated': False}],\n", + " 'UnitsAbbreviation': ''}]}" + ] + }, + "execution_count": 35, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "item = web_ids['CI-W3A05F1']['webid']\n", + "\n", + "requests.get(\n", + " f'https://pivision.votorantimcimentos.com/piwebapi/streamsets/recorded',\n", + " params={\n", + " 'webid': item,\n", + " 'startTime': '*-1d',\n", + " 'endtime': '*',\n", + " \"maxCount\": 1,\n", + " },\n", + " headers=headers\n", + ").json()" + ] + }, + { + "cell_type": "code", + "execution_count": 18, + "id": "f8c425e2", + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
variableCI-J3J01S1CI-J3P01T1ACI-J3P03S1CI-W3A05F1CI-W3A50A1CI-W3A50A2CI-W3A50A3CI-W3A50P1CI-W3A50T1CI-W3A55P1...CI-W3V33P1CI-W3W01A1CI-W3W01A2CI-W3W01A3CI-W3W01G1CI-W3W01P1CI-W3W01P2CI-W3W03I1CI-W3W03S1timestamp
087.999020238.000000100.442688272.8691000.0833382.636171589.246400-0.923784383.722400-18.167961...259.5655000.0082204.190174376.848267600.466900-3.1095980.22179863.6058301545.502932025-12-17 22:12:42
187.999020238.000000100.442688272.6492000.0833382.267340618.462100-0.771080381.468872-17.495136...263.0998540.0082202.611867435.797668600.466900-3.0301560.15352765.7857061545.502932025-12-17 22:41:54
286.997925228.200012100.442688264.2253720.0823583.254068444.800200-0.789923389.007800-16.590466...223.3463130.0859279.447197343.596000600.000000-2.4367670.35028969.8865201616.154912025-12-18 19:08:08
386.997925228.200012100.442688266.7141720.0823612.444070481.586060-0.551620390.612854-16.791473...240.6614690.0859279.324739462.062256600.000000-3.0301560.20861664.6597301628.228762025-12-18 19:23:08
491.991210139.299988100.442688280.3923340.0823611.835279436.144100-0.486019387.727000-17.495136...249.9351650.0596025.359922367.250300600.000000-3.2863160.27637259.9493371691.234622025-12-18 19:31:54
590.001220230.000000100.442688268.8931580.0836232.572147410.169100-0.929219395.898200-16.390797...241.3099670.0020004.262301340.296700880.778200-3.6710640.24807565.1953901609.201542025-12-18 19:34:28
690.001220230.000000100.442688269.3590000.0836232.636187417.679138-0.837243396.060272-16.390797...239.3646240.0020003.964616318.547300935.914368-3.4776270.24807563.2846381609.201542025-12-18 19:35:19
790.001220230.000000100.442688268.6519780.0836232.483914404.910522-0.906091396.222473-16.390797...239.3646240.0020004.766942314.721130903.813232-3.7976250.24807564.6019741645.863892025-12-18 19:39:04
890.001220230.000000100.442688270.8364260.0836232.748308403.442100-0.939606396.222473-16.390797...239.3646240.0013624.637836325.616100909.961060-3.6057070.24807562.8696441646.514282025-12-18 19:40:58
990.001220227.700012100.442688270.7156370.0897542.644230425.803000-1.007370396.384521-16.725641...239.0402220.0013615.083293349.286682949.923500-3.7029830.20841365.8502961647.255372025-12-18 20:12:23
1090.001220227.700012100.442688270.6614380.0788592.628080435.462860-0.962243396.384521-17.031452...239.3644410.1086905.151842353.761353949.961060-3.7337880.20741364.7887401644.680422025-12-19 03:26:44
1190.001220227.700012100.442688270.7251590.0788592.628080424.227722-0.827545396.384521-17.031452...239.3644410.1086904.830516353.761353893.074000-3.7337880.20741364.7887401644.680422025-12-19 03:32:37
1290.001220227.700012100.442688269.9770200.0788593.799854350.998047-0.901527395.614441-18.777557...226.5562740.1086905.567447312.856700887.664734-3.4452010.18896765.3761441644.680422025-12-19 03:59:20
1390.001220227.700012100.442688269.0980000.0788593.325234401.663940-0.960138392.380127-16.966602...235.8625180.0251952.443661305.058400963.424100-3.6543450.18896764.8246601392.283942025-12-19 04:07:07
1489.000120133.000000100.442688280.5959000.0788592.347593453.182526-0.409587389.169952-15.958172...238.0998540.0251955.168004417.250244852.529200-3.7029830.26809562.1951981645.463262025-12-19 04:27:35
1591.002320130.000000100.442688283.0378420.1095972.547736364.069400-0.737562394.961900-18.296043...226.2309000.0730104.334429338.391663793.501953-4.7276260.00395972.8045961718.755862025-12-21 14:12:47
\n", + "

16 rows × 33 columns

\n", + "
" + ], + "text/plain": [ + "variable CI-J3J01S1 CI-J3P01T1A CI-J3P03S1 CI-W3A05F1 CI-W3A50A1 \\\n", + "0 87.999020 238.000000 100.442688 272.869100 0.083338 \n", + "1 87.999020 238.000000 100.442688 272.649200 0.083338 \n", + "2 86.997925 228.200012 100.442688 264.225372 0.082358 \n", + "3 86.997925 228.200012 100.442688 266.714172 0.082361 \n", + "4 91.991210 139.299988 100.442688 280.392334 0.082361 \n", + "5 90.001220 230.000000 100.442688 268.893158 0.083623 \n", + "6 90.001220 230.000000 100.442688 269.359000 0.083623 \n", + "7 90.001220 230.000000 100.442688 268.651978 0.083623 \n", + "8 90.001220 230.000000 100.442688 270.836426 0.083623 \n", + "9 90.001220 227.700012 100.442688 270.715637 0.089754 \n", + "10 90.001220 227.700012 100.442688 270.661438 0.078859 \n", + "11 90.001220 227.700012 100.442688 270.725159 0.078859 \n", + "12 90.001220 227.700012 100.442688 269.977020 0.078859 \n", + "13 90.001220 227.700012 100.442688 269.098000 0.078859 \n", + "14 89.000120 133.000000 100.442688 280.595900 0.078859 \n", + "15 91.002320 130.000000 100.442688 283.037842 0.109597 \n", + "\n", + "variable CI-W3A50A2 CI-W3A50A3 CI-W3A50P1 CI-W3A50T1 CI-W3A55P1 ... \\\n", + "0 2.636171 589.246400 -0.923784 383.722400 -18.167961 ... \n", + "1 2.267340 618.462100 -0.771080 381.468872 -17.495136 ... \n", + "2 3.254068 444.800200 -0.789923 389.007800 -16.590466 ... \n", + "3 2.444070 481.586060 -0.551620 390.612854 -16.791473 ... \n", + "4 1.835279 436.144100 -0.486019 387.727000 -17.495136 ... \n", + "5 2.572147 410.169100 -0.929219 395.898200 -16.390797 ... \n", + "6 2.636187 417.679138 -0.837243 396.060272 -16.390797 ... \n", + "7 2.483914 404.910522 -0.906091 396.222473 -16.390797 ... \n", + "8 2.748308 403.442100 -0.939606 396.222473 -16.390797 ... \n", + "9 2.644230 425.803000 -1.007370 396.384521 -16.725641 ... \n", + "10 2.628080 435.462860 -0.962243 396.384521 -17.031452 ... \n", + "11 2.628080 424.227722 -0.827545 396.384521 -17.031452 ... \n", + "12 3.799854 350.998047 -0.901527 395.614441 -18.777557 ... \n", + "13 3.325234 401.663940 -0.960138 392.380127 -16.966602 ... \n", + "14 2.347593 453.182526 -0.409587 389.169952 -15.958172 ... \n", + "15 2.547736 364.069400 -0.737562 394.961900 -18.296043 ... \n", + "\n", + "variable CI-W3V33P1 CI-W3W01A1 CI-W3W01A2 CI-W3W01A3 CI-W3W01G1 \\\n", + "0 259.565500 0.008220 4.190174 376.848267 600.466900 \n", + "1 263.099854 0.008220 2.611867 435.797668 600.466900 \n", + "2 223.346313 0.085927 9.447197 343.596000 600.000000 \n", + "3 240.661469 0.085927 9.324739 462.062256 600.000000 \n", + "4 249.935165 0.059602 5.359922 367.250300 600.000000 \n", + "5 241.309967 0.002000 4.262301 340.296700 880.778200 \n", + "6 239.364624 0.002000 3.964616 318.547300 935.914368 \n", + "7 239.364624 0.002000 4.766942 314.721130 903.813232 \n", + "8 239.364624 0.001362 4.637836 325.616100 909.961060 \n", + "9 239.040222 0.001361 5.083293 349.286682 949.923500 \n", + "10 239.364441 0.108690 5.151842 353.761353 949.961060 \n", + "11 239.364441 0.108690 4.830516 353.761353 893.074000 \n", + "12 226.556274 0.108690 5.567447 312.856700 887.664734 \n", + "13 235.862518 0.025195 2.443661 305.058400 963.424100 \n", + "14 238.099854 0.025195 5.168004 417.250244 852.529200 \n", + "15 226.230900 0.073010 4.334429 338.391663 793.501953 \n", + "\n", + "variable CI-W3W01P1 CI-W3W01P2 CI-W3W03I1 CI-W3W03S1 timestamp \n", + "0 -3.109598 0.221798 63.605830 1545.50293 2025-12-17 22:12:42 \n", + "1 -3.030156 0.153527 65.785706 1545.50293 2025-12-17 22:41:54 \n", + "2 -2.436767 0.350289 69.886520 1616.15491 2025-12-18 19:08:08 \n", + "3 -3.030156 0.208616 64.659730 1628.22876 2025-12-18 19:23:08 \n", + "4 -3.286316 0.276372 59.949337 1691.23462 2025-12-18 19:31:54 \n", + "5 -3.671064 0.248075 65.195390 1609.20154 2025-12-18 19:34:28 \n", + "6 -3.477627 0.248075 63.284638 1609.20154 2025-12-18 19:35:19 \n", + "7 -3.797625 0.248075 64.601974 1645.86389 2025-12-18 19:39:04 \n", + "8 -3.605707 0.248075 62.869644 1646.51428 2025-12-18 19:40:58 \n", + "9 -3.702983 0.208413 65.850296 1647.25537 2025-12-18 20:12:23 \n", + "10 -3.733788 0.207413 64.788740 1644.68042 2025-12-19 03:26:44 \n", + "11 -3.733788 0.207413 64.788740 1644.68042 2025-12-19 03:32:37 \n", + "12 -3.445201 0.188967 65.376144 1644.68042 2025-12-19 03:59:20 \n", + "13 -3.654345 0.188967 64.824660 1392.28394 2025-12-19 04:07:07 \n", + "14 -3.702983 0.268095 62.195198 1645.46326 2025-12-19 04:27:35 \n", + "15 -4.727626 0.003959 72.804596 1718.75586 2025-12-21 14:12:47 \n", + "\n", + "[16 rows x 33 columns]" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "from pandas import read_csv, to_datetime\n", + "\n", + "df = read_csv('/home/grezewave/Downloads/laborious_data_202512220821.csv')\n", + "data = df.pivot(index='timestamp', columns='variable', values='value')\n", + "data['timestamp'] = data.index\n", + "\n", + "#Remove tz from timestamp\n", + "data['timestamp'] = to_datetime(data['timestamp'])\n", + "data['timestamp'] = data['timestamp'].dt.tz_localize(None)\n", + "\n", + "# Back to string and add \"\"\n", + "data['timestamp'] = data['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S')\n", + "data.reset_index(drop=True, inplace=True)\n", + "data.to_csv('VC-model-data.csv', index=False)\n", + "\n", + "display(data)" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.11.14" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/tests/activities/test_activities.py b/tests/activities/test_activities.py index bc31dee..2d3f5ab 100644 --- a/tests/activities/test_activities.py +++ b/tests/activities/test_activities.py @@ -3,6 +3,7 @@ from unittest.mock import ANY, MagicMock, patch from sientia_do.temporal.activities.postgres import Postgres from scouter.activities.activities import Activities +from scouter.activities.api import API from scouter.activities.gates import Gates from scouter.activities.mongodb import MongoDB from scouter.activities.redis import Redis @@ -12,9 +13,15 @@ from scouter.activities.redis import Redis @patch('scouter.activities.activities.Postgres.__init__') @patch('scouter.activities.activities.Redis.__init__') @patch('scouter.activities.activities.Gates.__init__') +@patch('scouter.activities.activities.API.__init__') @patch('scouter.activities.activities.MetricsController') def test___init__( - mock_metrics_controller, mock_gates_init, mock_redis_init, mock_postgres_init, mock_mongodb_init + mock_metrics_controller, + mock_api_init, + mock_gates_init, + mock_redis_init, + mock_postgres_init, + mock_mongodb_init, ): postgres_config = { 'host': 'localhost', @@ -33,6 +40,12 @@ def test___init__( 'database_name': 'test_database', } + api_config = { + 'base_url': 'https://api.example.com', + 'auth_type': 'bearer', + 'auth_token': 'test_token', + } + logger = MagicMock() notification_handler = MagicMock() @@ -40,6 +53,7 @@ def test___init__( postgres_config=postgres_config, redis_config=redis_config, mongodb_config=mongodb_config, + api_config=api_config, logger=logger, notification_handler=notification_handler, ) @@ -49,6 +63,7 @@ def test___init__( assert isinstance(activities, Redis) assert isinstance(activities, MongoDB) assert isinstance(activities, Gates) + assert isinstance(activities, API) mock_postgres_init.assert_called_once_with( ANY, @@ -91,20 +106,34 @@ def test___init__( metrics_controller=mock_metrics_controller.return_value, ) + mock_api_init.assert_called_once_with( + ANY, + base_url=api_config['base_url'], + auth_type=api_config['auth_type'], + auth_token=api_config['auth_token'], + logger=logger, + notification_handler=notification_handler, + metrics_controller=mock_metrics_controller.return_value, + ) + @patch('scouter.activities.activities.Postgres.__init__') @patch('scouter.activities.activities.Redis.__init__') @patch('scouter.activities.activities.Gates.__init__') @patch('scouter.activities.activities.MongoDB.__init__') +@patch('scouter.activities.activities.API.__init__') @patch('scouter.activities.activities.Postgres.close') @patch('scouter.activities.activities.MongoDB.close') @patch('scouter.activities.activities.Redis.close') @patch('scouter.activities.activities.Gates.close') +@patch('scouter.activities.activities.API.close') def test_shutdown( + mock_api_close, mock_gates_close, mock_redis_close, mock_mongodb_close, mock_postgres_close, + _mock_api_init, _mock_mongodb_init, _mock_gates_init, _mock_redis_init, @@ -127,6 +156,12 @@ def test_shutdown( 'database_name': 'test_database', } + api_config = { + 'base_url': 'https://api.example.com', + 'auth_type': 'bearer', + 'auth_token': 'test_token', + } + logger = MagicMock() notification_handler = MagicMock() @@ -134,6 +169,7 @@ def test_shutdown( postgres_config=postgres_config, redis_config=redis_config, mongodb_config=mongodb_config, + api_config=api_config, logger=logger, notification_handler=notification_handler, ) @@ -144,3 +180,4 @@ def test_shutdown( mock_mongodb_close.assert_called() mock_redis_close.assert_called() mock_gates_close.assert_called() + mock_api_close.assert_called() diff --git a/tests/activities/test_api.py b/tests/activities/test_api.py new file mode 100644 index 0000000..77810aa --- /dev/null +++ b/tests/activities/test_api.py @@ -0,0 +1,337 @@ +from unittest.mock import ANY, AsyncMock, MagicMock, patch + +import pandas as pd +import pytest +from sientia_do.notifications.models import NotificationLevel + +from scouter.activities.api import API + + +@pytest.fixture +@patch('scouter.activities.api.PIWebAPIClient') +def api_activity(mock_pi_web_api_client): + """Fixture to create an API activity instance with mocked dependencies.""" + logger = MagicMock() + notification_handler = MagicMock() + metrics_controller = MagicMock() + + activity = API( + base_url='https://pi.example.com', + auth_type='basic', + auth_token='test_token', + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + activity.logger = logger + activity.notification_handler = notification_handler + activity.metrics_controller = metrics_controller + activity.pod_id = 'test_pod_id' + + return activity + + +metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'pi_web_api_scouter', + } +} + + +@patch('scouter.activities.api.PIWebAPIClient') +def test_api_initialization(mock_pi_web_api_client): + """Test API activity initialization.""" + logger = MagicMock() + notification_handler = MagicMock() + metrics_controller = MagicMock() + + activity = API( + base_url='https://pi.example.com', + auth_type='basic', + auth_token='test_token', + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + mock_pi_web_api_client.assert_called_once_with( + base_url='https://pi.example.com', + auth_config={ + 'type': 'basic', + 'token': 'test_token', + }, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + + assert activity.pi_web_api_client is not None + + +@patch('scouter.activities.api.SientiaMonitoring') +def test_close(mock_sientia_monitoring, api_activity): + """Test close method.""" + api_activity.close() + api_activity.pi_web_api_client.close.assert_called_once() + mock_sientia_monitoring.shutdown.assert_called_once() + + +@pytest.mark.asyncio +async def test_get_tag_values_success(api_activity): + """Test get_tag_values with successful data retrieval.""" + # Setup test data + test_data = { + **metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': { + 'tag1': { + 'webid': 'webid1', + 'aggr_function': 'avg', + 'data_range': [0, 100], + }, + 'tag2': { + 'webid': 'webid2', + 'aggr_function': 'avg', + 'data_range': [0, 100], + }, + 'tag3': { + 'webid': 'webid3', + 'aggr_function': 'avg', + 'data_range': [0, 100], + }, + }, + 'period': '*-1d', + 'max_count': 10, + 'api_timeout': 30, + } + + # Mock DataFrame response + mock_df = pd.DataFrame( + { + 'timestamp': [ + '2023-01-01 12:00:00+0000', + '2023-01-01 12:01:00+0000', + '2023-01-01 12:02:00+0000', + ], + 'name': ['tag1', 'tag2', 'tag3'], + 'value': [10.5, 20.3, 30.7], + 'tag': ['webid1', 'webid2', 'webid3'], + } + ) + + mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) + + api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) + + # Execute + result = await api_activity.get_tag_values(test_data) + + # Verify + api_activity.pi_web_api_client.get_latest_values_df.assert_called_once_with( + endpoint='/streamsets/recorded', + web_ids={ + 'tag1': {'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100]}, + 'tag2': {'webid': 'webid2', 'aggr_function': 'avg', 'data_range': [0, 100]}, + 'tag3': {'webid': 'webid3', 'aggr_function': 'avg', 'data_range': [0, 100]}, + }, + start_time='*-1d', + max_count=10, + metadata=metadata['metadata'], + timeout=30, + ) + + assert len(result) == 3 + assert result[0]['name'] == 'tag1' + assert result[0]['value'] == 10.5 + assert result[1]['name'] == 'tag2' + assert result[2]['name'] == 'tag3' + assert result[0]['timestamp'] == '2023-01-01 12:02:00+0000' + assert result[1]['timestamp'] == '2023-01-01 12:02:00+0000' + assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000' + + +@pytest.mark.asyncio +async def test_get_tag_values_with_default_max_count(api_activity): + """Test get_tag_values with default max_count value.""" + # Setup test data without max_count + test_data = { + **metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': { + 'tag1': { + 'webid': 'webid1', + 'aggr_function': 'avg', + 'data_range': [0, 100], + } + }, + 'period': '*-1h', + 'api_timeout': 15, + } + + mock_df = pd.DataFrame( + { + 'timestamp': ['2023-01-01 12:00:00+0000'], + 'name': ['tag1'], + 'value': [42.0], + 'tag': ['webid1'], + } + ) + + mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) + + api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) + + # Execute + result = await api_activity.get_tag_values(test_data) + + # Verify default max_count is 1 + api_activity.pi_web_api_client.get_latest_values_df.assert_called_once_with( + endpoint='/streamsets/recorded', + web_ids={'tag1': {'webid': 'webid1', 'aggr_function': 'avg', 'data_range': [0, 100]}}, + start_time='*-1h', + max_count=1, + metadata=metadata['metadata'], + timeout=15, + ) + + assert len(result) == 1 + + +@pytest.mark.asyncio +async def test_get_tag_values_with_none_webids(api_activity): + """Test get_tag_values with some None WebIds.""" + # Setup test data with None values + test_data = { + **metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': { + 'tag1': { + 'webid': 'webid1', + 'aggr_function': 'avg', + 'data_range': [0, 100], + }, + 'tag2': None, + 'tag3': { + 'webid': 'webid3', + 'aggr_function': 'max', + 'data_range': [0, 200], + }, + }, + 'period': '*-1h', + 'max_count': 5, + 'api_timeout': 20, + } + + mock_df = pd.DataFrame( + { + 'timestamp': ['2023-01-01 12:00:00+0000', '2023-01-01 12:01:00+0000'], + 'name': ['tag1', 'tag3'], + 'value': [10.5, 30.7], + 'tag': ['webid1', 'webid3'], + } + ) + + mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) + + api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) + + # Execute + result = await api_activity.get_tag_values(test_data) + + # Verify - should only query non-None WebIds + assert len(result) == 2 + assert all(r['name'] in ['tag1', 'tag3'] for r in result) + + +@pytest.mark.asyncio +async def test_get_tag_values_api_error(api_activity): + """Test get_tag_values when PI Web API client raises an error and sends notification.""" + # Setup test data + test_data = { + **metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': { + 'tag1': { + 'webid': 'webid1', + 'aggr_function': 'avg', + 'data_range': [0, 100], + } + }, + 'period': '*-1d', + 'max_count': 1, + 'api_timeout': 30, + } + + # Mock API error + api_activity.pi_web_api_client.get_latest_values_df = AsyncMock( + side_effect=Exception('PI Web API connection error') + ) + api_activity.send_notification_async = AsyncMock() + + # Execute and verify exception is raised + with pytest.raises(Exception) as exc_info: + await api_activity.get_tag_values(test_data) + + assert str(exc_info.value) == 'PI Web API connection error' + + # Verify notification was sent + api_activity.send_notification_async.assert_called_once_with( + metadata=metadata['metadata'], + notification_id='PI_WEB_API_REQUEST_ERROR', + message='Error getting tag values from PI Web API: PI Web API connection error', + block='get_tag_values', + level=NotificationLevel.ERROR, + attachment_content=ANY, + ) + + +@pytest.mark.asyncio +async def test_get_tag_values_with_nan_values(api_activity): + """Test get_tag_values handling NaN values in the DataFrame.""" + # Setup test data + test_data = { + **metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': { + 'tag1': { + 'webid': 'webid1', + 'aggr_function': 'avg', + 'data_range': [0, 100], + }, + 'tag2': { + 'webid': 'webid2', + 'aggr_function': 'avg', + 'data_range': [0, 100], + }, + }, + 'period': '*-1d', + 'max_count': 1, + 'api_timeout': 30, + } + + # Mock DataFrame with NaN values + mock_df = pd.DataFrame( + { + 'timestamp': ['2023-01-01 12:00:00+0000', '2023-01-01 12:00:00+0000'], + 'name': ['tag1', 'tag2'], + 'value': [10.0, float('nan')], + 'tag': ['webid1', 'webid2'], + } + ) + + mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp']) + + api_activity.pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df) + + # Execute + result = await api_activity.get_tag_values(test_data) + + # Verify + assert len(result) == 2 + assert result[0]['value'] == 10.0 + # NaN should be preserved in the result + assert pd.isna(result[1]['value']) diff --git a/tests/utils/clients/__init__.py b/tests/utils/clients/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/utils/clients/test_pi_web_api_client.py b/tests/utils/clients/test_pi_web_api_client.py new file mode 100644 index 0000000..7c897e3 --- /dev/null +++ b/tests/utils/clients/test_pi_web_api_client.py @@ -0,0 +1,560 @@ +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pandas as pd +import pycurl +import pytest + +from scouter.utils.clients.pi_web_api_client import PIMSRequestError, PIWebAPIClient + + +@pytest.fixture +def mock_logger(): + return MagicMock() + + +@pytest.fixture +def mock_notification_handler(): + return AsyncMock() + + +@pytest.fixture +def mock_metrics_controller(): + return AsyncMock() + + +@pytest.fixture +def auth_config_basic(): + return {'type': 'basic', 'token': 'test_token_123'} + + +@pytest.fixture +def auth_config_bearer(): + return {'type': 'bearer', 'token': 'bearer_token_456'} + + +@pytest.fixture +def pi_client(mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic): + return PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=auth_config_basic, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + +def test_init_with_basic_auth( + mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic +): + """Test initialization with basic authentication""" + client = PIWebAPIClient( + base_url='https://pi.example.com/', + auth_config=auth_config_basic, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert client.base_url == 'https://pi.example.com' + assert client.auth_config['type'] == 'basic' + assert client.headers['Authorization'] == 'Basic test_token_123' + assert client.headers['Content-Type'] == 'application/json' + assert client.headers['Accept'] == 'application/json' + mock_logger.info.assert_called_with('Authenticating with basic authentication') + + +def test_init_with_bearer_auth( + mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_bearer +): + """Test initialization with bearer authentication""" + client = PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=auth_config_bearer, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert client.base_url == 'https://pi.example.com' + assert client.auth_config['type'] == 'bearer' + assert client.headers['Authorization'] == 'Bearer bearer_token_456' + mock_logger.info.assert_called_with('Authenticating with bearer authentication') + + +def test_init_with_custom_headers( + mock_logger, mock_notification_handler, mock_metrics_controller, auth_config_basic +): + """Test initialization with custom headers""" + custom_headers = { + 'Content-Type': 'application/xml', + 'Custom-Header': 'custom_value', + } + + client = PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=auth_config_basic, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + headers_config=custom_headers, + ) + + assert client.headers['Content-Type'] == 'application/xml' + assert client.headers['Custom-Header'] == 'custom_value' + assert client.headers['Authorization'] == 'Basic test_token_123' + + +def test_authenticate_invalid_type(mock_logger, mock_notification_handler, mock_metrics_controller): + """Test that invalid authentication type raises ValueError""" + invalid_auth_config = {'type': 'invalid', 'token': 'test_token'} + + with pytest.raises(ValueError) as exc_info: + PIWebAPIClient( + base_url='https://pi.example.com', + auth_config=invalid_auth_config, + logger=mock_logger, + notification_handler=mock_notification_handler, + metrics_controller=mock_metrics_controller, + ) + + assert 'Invalid authentication type: invalid' in str(exc_info.value) + + +@patch('scouter.utils.clients.pi_web_api_client.SientiaMonitoring.shutdown') +def test_close(mock_shutdown, pi_client): + """Test close method calls shutdown""" + pi_client.close() + + mock_shutdown.assert_called_once() + + +def test_to_clean_timestamp(pi_client): + """Test timestamp cleaning and normalization""" + timestamps = pd.Series( + [ + '2025-01-15T10:30:45.123456Z', + '2025-01-15T10:30:46.789012Z', + '2025-01-15T10:30:47.999999Z', + ] + ) + + result = pi_client._to_clean_timestamp(timestamps) + + assert isinstance(result, pd.Series) + assert result.dtype == 'datetime64[ns, UTC]' + # Verify microseconds are floored to seconds + assert result[0] == pd.Timestamp('2025-01-15T10:30:45Z') + assert result[1] == pd.Timestamp('2025-01-15T10:30:46Z') + assert result[2] == pd.Timestamp('2025-01-15T10:30:47Z') + + +def test_to_clean_timestamp_with_invalid_values(pi_client): + """Test timestamp cleaning with invalid values returns NaT""" + timestamps = pd.Series(['invalid', 'not_a_date', '2025-01-15T10:30:45Z']) + + result = pi_client._to_clean_timestamp(timestamps) + + assert pd.isna(result[0]) + assert pd.isna(result[1]) + assert result[2] == pd.Timestamp('2025-01-15T10:30:45Z') + + +def test_extract_numeric_with_float(pi_client): + """Test extracting numeric value from float""" + result = pi_client._extract_numeric(42.5) + + assert result == 42.5 + + +def test_extract_numeric_with_int(pi_client): + """Test extracting numeric value from int""" + result = pi_client._extract_numeric(42) + + assert result == 42.0 + + +def test_extract_numeric_with_string(pi_client): + """Test extracting numeric value from string""" + result = pi_client._extract_numeric('123.45') + + assert result == 123.45 + + +def test_extract_numeric_with_dict(pi_client): + """Test extracting numeric value from dictionary""" + result = pi_client._extract_numeric({'Value': 99.9}) + + assert result == 99.9 + + +def test_extract_numeric_with_invalid_value(pi_client): + """Test extracting numeric value from invalid value returns None/NaN""" + result = pi_client._extract_numeric('invalid_number') + + assert pd.isna(result) + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_success(mock_curl_class, pi_client): + """Test successful GET request with JSON response""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'status': 'success', 'data': [1, 2, 3]} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + result = await pi_client._curl_get_json('https://pi.example.com/api/test') + + assert result == response_data + mock_curl.setopt.assert_any_call(pycurl.TIMEOUT, 30) + mock_curl.perform.assert_called_once() + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_with_params(mock_curl_class, pi_client): + """Test GET request with query parameters""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'result': 'ok'} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + params = [('key1', 'value1'), ('key2', 'value2')] + result = await pi_client._curl_get_json('https://pi.example.com/api', params=params) + + assert result == response_data + # Verify URL includes query parameters + set_url_call = [call for call in mock_curl.setopt.call_args_list if call[0][0] == pycurl.URL][0] + assert b'key1=value1' in set_url_call[0][1] + assert b'key2=value2' in set_url_call[0][1] + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_http_error(mock_curl_class, pi_client): + """Test GET request with HTTP error response""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + error_response = b'{"error": "Not found"}' + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(error_response) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 404 + + with pytest.raises(PIMSRequestError) as exc_info: + await pi_client._curl_get_json('https://pi.example.com/api/notfound') + + assert 'HTTP 404' in str(exc_info.value) + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_connection_error(mock_curl_class, pi_client): + """Test GET request with connection error""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + mock_curl.perform.side_effect = pycurl.error('Connection failed') + + with pytest.raises(PIMSRequestError) as exc_info: + await pi_client._curl_get_json('https://pi.example.com/api/test') + + assert 'Connection error' in str(exc_info.value) + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_invalid_json(mock_curl_class, pi_client): + """Test GET request with invalid JSON response""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + invalid_json = b'This is not valid JSON' + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(invalid_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + with pytest.raises(PIMSRequestError) as exc_info: + await pi_client._curl_get_json('https://pi.example.com/api/test') + + assert 'Error decoding JSON response' in str(exc_info.value) + mock_curl.close.assert_called_once() + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_with_custom_timeout(mock_curl_class, pi_client): + """Test GET request with custom timeout""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'status': 'ok'} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + await pi_client._curl_get_json('https://pi.example.com/api/test', timeout=60) + + mock_curl.setopt.assert_any_call(pycurl.TIMEOUT, 60) + + +@pytest.mark.asyncio +@patch('scouter.utils.clients.pi_web_api_client.pycurl.Curl') +async def test_curl_get_json_without_ssl_verify(mock_curl_class, pi_client): + """Test GET request with SSL verification disabled""" + mock_curl = MagicMock() + mock_curl_class.return_value = mock_curl + + response_data = {'status': 'ok'} + response_json = json.dumps(response_data).encode('utf-8') + + def mock_perform(): + buffer = mock_curl.setopt.call_args_list[1][0][1] + buffer.write(response_json) + + mock_curl.perform.side_effect = mock_perform + mock_curl.getinfo.return_value = 200 + + await pi_client._curl_get_json('https://pi.example.com/api/test', verify=False) + + mock_curl.setopt.assert_any_call(pycurl.SSL_VERIFYPEER, 0) + mock_curl.setopt.assert_any_call(pycurl.SSL_VERIFYHOST, 0) + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_success(mock_curl_get_json, pi_client): + """Test successful retrieval of latest values""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, + {'Timestamp': '2025-01-15T10:31:00Z', 'Value': 43.0}, + ], + }, + { + 'Name': 'tag2', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 100.0}, + ], + }, + ] + } + + web_ids = { + 'tag1': {'webid': 'webid1'}, + 'tag2': {'webid': 'webid2'}, + } + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + start_time='*-1d', + end_time='*', + max_count=10, + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 3 + assert list(result.columns) == ['timestamp', 'name', 'value', 'tag'] + assert result['name'].tolist() == ['tag1', 'tag1', 'tag2'] + assert result['value'].tolist() == [42.5, 43.0, 100.0] + + mock_curl_get_json.assert_called_once() + call_args = mock_curl_get_json.call_args + assert call_args[1]['url'] == 'https://pi.example.com/streamsets/recorded' + assert call_args[1]['timeout'] == 30 + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_with_custom_params(mock_curl_get_json, pi_client): + """Test get_latest_values_df with custom parameters""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, + ], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + metadata = {'model_id': 'test_model'} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + start_time='*-7d', + end_time='*-1d', + max_count=100, + timeout=60, + metadata=metadata, + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 1 + + mock_curl_get_json.assert_called_once() + call_args = mock_curl_get_json.call_args + params = call_args[1]['params'] + + # Verify parameters + assert ('startTime', '*-7d') in params + assert ('endtime', '*-1d') in params + assert ('maxCount', '100') in params + assert call_args[1]['timeout'] == 60 + assert call_args[1]['metadata'] == metadata + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_empty_response(mock_curl_get_json, pi_client): + """Test get_latest_values_df with empty response""" + mock_curl_get_json.return_value = {'Items': []} + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 0 + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_no_items_in_tag(mock_curl_get_json, pi_client): + """Test get_latest_values_df when tag has no items""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 0 + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_with_missing_timestamp(mock_curl_get_json, pi_client): + """Test get_latest_values_df filters out items with missing timestamp""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': 42.5}, + {'Value': 43.0}, # Missing Timestamp + {'Timestamp': None, 'Value': 44.0}, # None Timestamp + ], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 1 # Only the first item should be included + assert result['value'].tolist() == [42.5] + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_with_nested_value(mock_curl_get_json, pi_client): + """Test get_latest_values_df with nested value extraction""" + mock_curl_get_json.return_value = { + 'Items': [ + { + 'Name': 'tag1', + 'Items': [ + {'Timestamp': '2025-01-15T10:30:00Z', 'Value': {'Value': 42.5}}, + ], + } + ] + } + + web_ids = {'tag1': {'webid': 'webid1'}} + + result = await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + assert isinstance(result, pd.DataFrame) + assert len(result) == 1 + assert result['value'].tolist() == [42.5] + + +@pytest.mark.asyncio +@patch.object(PIWebAPIClient, '_curl_get_json', new_callable=AsyncMock) +async def test_get_latest_values_df_default_max_count(mock_curl_get_json, pi_client): + """Test get_latest_values_df uses default max_count of 1""" + mock_curl_get_json.return_value = {'Items': []} + + web_ids = {'tag1': {'webid': 'webid1'}} + + await pi_client.get_latest_values_df( + web_ids=web_ids, + endpoint='/streamsets/recorded', + ) + + call_args = mock_curl_get_json.call_args + params = call_args[1]['params'] + + assert ('maxCount', '1') in params diff --git a/tests/utils/test_connectors_config.py b/tests/utils/test_connectors_config.py index d57b335..be7ab70 100644 --- a/tests/utils/test_connectors_config.py +++ b/tests/utils/test_connectors_config.py @@ -4,6 +4,7 @@ from unittest.mock import patch import pytest from scouter.utils.connectors_config import ( + build_api_config, build_druid_config, build_kafka_config, build_mongodb_config, @@ -154,6 +155,38 @@ def test_build_mongodb_config_with_env_vars(): } +@pytest.mark.usefixtures('mock_env_vars') +def test_build_api_config_defaults(): + """Test that build_api_config returns default values when no env vars are set""" + config = build_api_config() + + assert config == { + 'base_url': 'https://pi.example.com', + 'auth_type': 'basic', + 'auth_token': None, + } + + +@pytest.mark.usefixtures('mock_env_vars') +def test_build_api_config_with_env_vars(): + """Test that build_api_config uses env vars when set""" + with patch.dict( + os.environ, + { + 'PI_WEB_API_BASE_URL': 'https://api.production.com', + 'PI_WEB_API_AUTH_TYPE': 'bearer', + 'PI_WEB_API_AUTH_TOKEN': 'secret_token_123', + }, + ): + config = build_api_config() + + assert config == { + 'base_url': 'https://api.production.com', + 'auth_type': 'bearer', + 'auth_token': 'secret_token_123', + } + + def test_build_druid_config_defaults(): """Test that build_druid_config returns default values when no env vars are set""" config = build_druid_config() diff --git a/tests/workflow/sub_workflows/test_core_scouter.py b/tests/workflow/sub_workflows/test_core_scouter.py index 259b721..608985e 100644 --- a/tests/workflow/sub_workflows/test_core_scouter.py +++ b/tests/workflow/sub_workflows/test_core_scouter.py @@ -20,6 +20,11 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter): 'grouped_data', 'held_data', ] + mock_workflow.execute_activity_method.side_effect = [ + {'affected_rows': 10}, # export_data_to_postgres + None, # write_metrics + None, # store_data_package + ] await core_scouter.run( input_data={ 'metadata': { @@ -113,6 +118,22 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter): 'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ, }, + 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + ] + ) + + mock_workflow.execute_activity_method.assert_has_calls( + [ + call( + Activities.write_metrics, + { + **expected_metadata, + 'tag_values': 'held_data', }, retry_policy=ANY, start_to_close_timeout=ANY, @@ -222,3 +243,161 @@ async def test_core_scouter_workflow_with_empty_data(mock_workflow, core_scouter ) assert mock_workflow.execute_local_activity_method.call_count == 3 + mock_workflow.execute_activity_method.assert_not_called() + + +@pytest.mark.asyncio +@patch('scouter.workflow.sub_workflows.core_scouter.workflow', new_callable=AsyncMock) +async def test_core_scouter_workflow_with_zero_affected_rows(mock_workflow, core_scouter): + """ + Test that workflow stops after export when no rows are affected + """ + mock_workflow.execute_local_activity_method.side_effect = [ + 'filtered_data', + 'grouped_data', + 'held_data', + ] + mock_workflow.execute_activity_method.return_value = {'affected_rows': 0} + + await core_scouter.run( + input_data={ + 'metadata': { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'test_workflow', + } + }, + 'workflow_name': 'test_workflow', + 'schedule_name': 'test_schedule', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'data': 'test_data', + 'trigger_laborious': False, + 'filters': {'test_filter': 'test_value'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + 'model_tags': {}, + 'debug_data_package': True, + 'fill_missing_tags': False, + } + ) + + expected_metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'test_workflow', + } + } + + mock_workflow.execute_activity_method.assert_called_once_with( + Activities.export_data_to_postgres, + { + **expected_metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'data': 'held_data', + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT_WITH_TZ, + }, + 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + + +@pytest.mark.asyncio +@patch('scouter.workflow.sub_workflows.core_scouter.workflow', new_callable=AsyncMock) +async def test_core_scouter_workflow_without_debug_data_package(mock_workflow, core_scouter): + """ + Test that store_data_package is not called when debug_data_package is False + """ + mock_workflow.execute_local_activity_method.side_effect = [ + 'filtered_data', + 'grouped_data', + 'held_data', + ] + mock_workflow.execute_activity_method.side_effect = [ + {'affected_rows': 5}, + None, + ] + + await core_scouter.run( + input_data={ + 'metadata': { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'test_workflow', + } + }, + 'workflow_name': 'test_workflow', + 'schedule_name': 'test_schedule', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'data': 'test_data', + 'trigger_laborious': False, + 'filters': {'test_filter': 'test_value'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + 'model_tags': {}, + 'debug_data_package': False, + 'fill_missing_tags': False, + } + ) + + expected_metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'test_workflow', + } + } + + mock_workflow.execute_activity_method.assert_has_calls( + [ + call( + Activities.export_data_to_postgres, + { + **expected_metadata, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'data': 'held_data', + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT_WITH_TZ, + }, + 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + ] + ) + + mock_workflow.execute_activity_method.assert_has_calls( + [ + call( + Activities.write_metrics, + { + **expected_metadata, + 'tag_values': 'held_data', + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + ] + ) + + assert mock_workflow.execute_activity_method.call_count == 2 diff --git a/tests/workflow/test_pi_web_api_scouter.py b/tests/workflow/test_pi_web_api_scouter.py new file mode 100644 index 0000000..4f7f381 --- /dev/null +++ b/tests/workflow/test_pi_web_api_scouter.py @@ -0,0 +1,133 @@ +from unittest.mock import ANY, AsyncMock, patch + +from pytest import fixture, mark + +from scouter.activities.activities import Activities +from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter + + +@fixture +def pi_web_api_scouter(): + return PIWebAPIScouter() + + +@mark.asyncio +@patch('scouter.workflow.pi_web_api_scouter.workflow', new_callable=AsyncMock) +async def test_pi_web_api_scouter_workflow(mock_workflow, pi_web_api_scouter): + mock_workflow.execute_local_activity_method.return_value = 'test_data' + await pi_web_api_scouter.run( + input_data={ + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'schedule_name': 'test_schedule', + 'model_tags': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'trigger_laborious': True, + 'filters': {'quality': 'good'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + 'pi_web_api_query': { + 'endpoint': '/streamsets/recorded', + 'period': '*-1d', + 'max_count': 10, + 'api_timeout': 30, + }, + } + ) + + expected_metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'scouter', + } + } + + mock_workflow.execute_local_activity_method.assert_called_once_with( + Activities.get_tag_values, + { + **expected_metadata, + 'endpoint': '/streamsets/recorded', + 'web_ids': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'period': '*-1d', + 'api_timeout': 30, + 'max_count': 10, + }, + start_to_close_timeout=ANY, + retry_policy=ANY, + ) + + mock_workflow.execute_child_workflow.assert_called_once_with( + 'subworkflow.core_scouter', + { + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'schedule_name': 'test_schedule', + 'model_tags': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'trigger_laborious': True, + 'filters': {'quality': 'good'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + 'workflow_name': 'scouter', + 'data': 'test_data', + 'metadata': expected_metadata, + 'pi_web_api_query': { + 'endpoint': '/streamsets/recorded', + 'period': '*-1d', + 'max_count': 10, + 'api_timeout': 30, + }, + }, + ) + + +@mark.asyncio +@patch('scouter.workflow.pi_web_api_scouter.workflow', new_callable=AsyncMock) +async def test_pi_web_api_scouter_workflow_empty(mock_workflow, pi_web_api_scouter): + mock_workflow.execute_local_activity_method.return_value = [] + await pi_web_api_scouter.run( + input_data={ + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'schedule_name': 'test_schedule', + 'model_tags': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'trigger_laborious': True, + 'filters': {'quality': 'good'}, + 'schema': 'test_schema', + 'table_name': 'test_table', + 'retention_time': 3600, + 'pi_web_api_query': { + 'endpoint': '/streamsets/recorded', + 'period': '*-1d', + 'max_count': 1, + 'api_timeout': 30, + }, + } + ) + + expected_metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'schedule_name': 'test_schedule', + 'workflow_name': 'scouter', + } + } + + mock_workflow.execute_local_activity_method.assert_called_once_with( + Activities.get_tag_values, + { + **expected_metadata, + 'web_ids': {'tag1': 'webid1', 'tag2': 'webid2'}, + 'period': '*-1d', + 'api_timeout': 30, + 'max_count': 1, + 'endpoint': '/streamsets/recorded', + }, + start_to_close_timeout=ANY, + retry_policy=ANY, + ) + + mock_workflow.execute_child_workflow.assert_not_called() diff --git a/values.yaml b/values.yaml index 56d9b71..8c690ae 100644 --- a/values.yaml +++ b/values.yaml @@ -163,7 +163,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git" - name: GITHUB_BRANCH - value: "release/SIENTIAPDE-1441" + value: "feature/SIENTIAPDE-1445" - name: PYTHON_APP value: "scouter.worker.worker" @@ -219,6 +219,16 @@ env: - name: MONGODB_DATABASE value: "sientia" + - name: PI_WEB_API_BASE_URL + value: "https://pivision.votorantimcimentos.com/piwebapi" + - name: PI_WEB_API_AUTH_TYPE + value: "basic" + - name: PI_WEB_API_AUTH_TOKEN + valueFrom: + secretKeyRef: + name: pi-web-api-auth-token + key: token + - name: PYPI_SERVER value: "http://library-distribution-server.library.svc.cluster.local:5000" @@ -260,4 +270,8 @@ ssh: # kubectl create secret generic git-ssh-key-sientia-scouter-worker \ # --namespace sientia \ # --from-file=ssh-privatekey=git_key \ -# --type=kubernetes.io/ssh-auth \ No newline at end of file +# --type=kubernetes.io/ssh-auth + +# kubectl create secret generic pi-web-api-auth-token \ +# --namespace sientia \ +# --from-literal=token=your-token-here \ No newline at end of file