Merge pull request #27 from Aignosi/feature/SIENTIAPDE-1445

Enhance PI Web API Integration, Refactor Worker Setup, and Update Dependencies
This commit is contained in:
vitor-aignosi
2025-12-22 15:22:53 -03:00
committed by GitHub
20 changed files with 2962 additions and 10 deletions

View File

@@ -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

View File

@@ -3,5 +3,6 @@ psycopg2-binary
sqlalchemy
redis
pymongo
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.7.1
prometheus-client
pycurl

View File

@@ -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)

139
scouter/activities/api.py Normal file
View File

@@ -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')

View File

@@ -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,
)

View File

@@ -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

View File

@@ -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.

View File

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

View File

@@ -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 = []

View File

@@ -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)

View File

@@ -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,
{

984
tests.ipynb Normal file
View File

@@ -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": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th>variable</th>\n",
" <th>CI-J3J01S1</th>\n",
" <th>CI-J3P01T1A</th>\n",
" <th>CI-J3P03S1</th>\n",
" <th>CI-W3A05F1</th>\n",
" <th>CI-W3A50A1</th>\n",
" <th>CI-W3A50A2</th>\n",
" <th>CI-W3A50A3</th>\n",
" <th>CI-W3A50P1</th>\n",
" <th>CI-W3A50T1</th>\n",
" <th>CI-W3A55P1</th>\n",
" <th>...</th>\n",
" <th>CI-W3V33P1</th>\n",
" <th>CI-W3W01A1</th>\n",
" <th>CI-W3W01A2</th>\n",
" <th>CI-W3W01A3</th>\n",
" <th>CI-W3W01G1</th>\n",
" <th>CI-W3W01P1</th>\n",
" <th>CI-W3W01P2</th>\n",
" <th>CI-W3W03I1</th>\n",
" <th>CI-W3W03S1</th>\n",
" <th>timestamp</th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>0</th>\n",
" <td>87.999020</td>\n",
" <td>238.000000</td>\n",
" <td>100.442688</td>\n",
" <td>272.869100</td>\n",
" <td>0.083338</td>\n",
" <td>2.636171</td>\n",
" <td>589.246400</td>\n",
" <td>-0.923784</td>\n",
" <td>383.722400</td>\n",
" <td>-18.167961</td>\n",
" <td>...</td>\n",
" <td>259.565500</td>\n",
" <td>0.008220</td>\n",
" <td>4.190174</td>\n",
" <td>376.848267</td>\n",
" <td>600.466900</td>\n",
" <td>-3.109598</td>\n",
" <td>0.221798</td>\n",
" <td>63.605830</td>\n",
" <td>1545.50293</td>\n",
" <td>2025-12-17 22:12:42</td>\n",
" </tr>\n",
" <tr>\n",
" <th>1</th>\n",
" <td>87.999020</td>\n",
" <td>238.000000</td>\n",
" <td>100.442688</td>\n",
" <td>272.649200</td>\n",
" <td>0.083338</td>\n",
" <td>2.267340</td>\n",
" <td>618.462100</td>\n",
" <td>-0.771080</td>\n",
" <td>381.468872</td>\n",
" <td>-17.495136</td>\n",
" <td>...</td>\n",
" <td>263.099854</td>\n",
" <td>0.008220</td>\n",
" <td>2.611867</td>\n",
" <td>435.797668</td>\n",
" <td>600.466900</td>\n",
" <td>-3.030156</td>\n",
" <td>0.153527</td>\n",
" <td>65.785706</td>\n",
" <td>1545.50293</td>\n",
" <td>2025-12-17 22:41:54</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2</th>\n",
" <td>86.997925</td>\n",
" <td>228.200012</td>\n",
" <td>100.442688</td>\n",
" <td>264.225372</td>\n",
" <td>0.082358</td>\n",
" <td>3.254068</td>\n",
" <td>444.800200</td>\n",
" <td>-0.789923</td>\n",
" <td>389.007800</td>\n",
" <td>-16.590466</td>\n",
" <td>...</td>\n",
" <td>223.346313</td>\n",
" <td>0.085927</td>\n",
" <td>9.447197</td>\n",
" <td>343.596000</td>\n",
" <td>600.000000</td>\n",
" <td>-2.436767</td>\n",
" <td>0.350289</td>\n",
" <td>69.886520</td>\n",
" <td>1616.15491</td>\n",
" <td>2025-12-18 19:08:08</td>\n",
" </tr>\n",
" <tr>\n",
" <th>3</th>\n",
" <td>86.997925</td>\n",
" <td>228.200012</td>\n",
" <td>100.442688</td>\n",
" <td>266.714172</td>\n",
" <td>0.082361</td>\n",
" <td>2.444070</td>\n",
" <td>481.586060</td>\n",
" <td>-0.551620</td>\n",
" <td>390.612854</td>\n",
" <td>-16.791473</td>\n",
" <td>...</td>\n",
" <td>240.661469</td>\n",
" <td>0.085927</td>\n",
" <td>9.324739</td>\n",
" <td>462.062256</td>\n",
" <td>600.000000</td>\n",
" <td>-3.030156</td>\n",
" <td>0.208616</td>\n",
" <td>64.659730</td>\n",
" <td>1628.22876</td>\n",
" <td>2025-12-18 19:23:08</td>\n",
" </tr>\n",
" <tr>\n",
" <th>4</th>\n",
" <td>91.991210</td>\n",
" <td>139.299988</td>\n",
" <td>100.442688</td>\n",
" <td>280.392334</td>\n",
" <td>0.082361</td>\n",
" <td>1.835279</td>\n",
" <td>436.144100</td>\n",
" <td>-0.486019</td>\n",
" <td>387.727000</td>\n",
" <td>-17.495136</td>\n",
" <td>...</td>\n",
" <td>249.935165</td>\n",
" <td>0.059602</td>\n",
" <td>5.359922</td>\n",
" <td>367.250300</td>\n",
" <td>600.000000</td>\n",
" <td>-3.286316</td>\n",
" <td>0.276372</td>\n",
" <td>59.949337</td>\n",
" <td>1691.23462</td>\n",
" <td>2025-12-18 19:31:54</td>\n",
" </tr>\n",
" <tr>\n",
" <th>5</th>\n",
" <td>90.001220</td>\n",
" <td>230.000000</td>\n",
" <td>100.442688</td>\n",
" <td>268.893158</td>\n",
" <td>0.083623</td>\n",
" <td>2.572147</td>\n",
" <td>410.169100</td>\n",
" <td>-0.929219</td>\n",
" <td>395.898200</td>\n",
" <td>-16.390797</td>\n",
" <td>...</td>\n",
" <td>241.309967</td>\n",
" <td>0.002000</td>\n",
" <td>4.262301</td>\n",
" <td>340.296700</td>\n",
" <td>880.778200</td>\n",
" <td>-3.671064</td>\n",
" <td>0.248075</td>\n",
" <td>65.195390</td>\n",
" <td>1609.20154</td>\n",
" <td>2025-12-18 19:34:28</td>\n",
" </tr>\n",
" <tr>\n",
" <th>6</th>\n",
" <td>90.001220</td>\n",
" <td>230.000000</td>\n",
" <td>100.442688</td>\n",
" <td>269.359000</td>\n",
" <td>0.083623</td>\n",
" <td>2.636187</td>\n",
" <td>417.679138</td>\n",
" <td>-0.837243</td>\n",
" <td>396.060272</td>\n",
" <td>-16.390797</td>\n",
" <td>...</td>\n",
" <td>239.364624</td>\n",
" <td>0.002000</td>\n",
" <td>3.964616</td>\n",
" <td>318.547300</td>\n",
" <td>935.914368</td>\n",
" <td>-3.477627</td>\n",
" <td>0.248075</td>\n",
" <td>63.284638</td>\n",
" <td>1609.20154</td>\n",
" <td>2025-12-18 19:35:19</td>\n",
" </tr>\n",
" <tr>\n",
" <th>7</th>\n",
" <td>90.001220</td>\n",
" <td>230.000000</td>\n",
" <td>100.442688</td>\n",
" <td>268.651978</td>\n",
" <td>0.083623</td>\n",
" <td>2.483914</td>\n",
" <td>404.910522</td>\n",
" <td>-0.906091</td>\n",
" <td>396.222473</td>\n",
" <td>-16.390797</td>\n",
" <td>...</td>\n",
" <td>239.364624</td>\n",
" <td>0.002000</td>\n",
" <td>4.766942</td>\n",
" <td>314.721130</td>\n",
" <td>903.813232</td>\n",
" <td>-3.797625</td>\n",
" <td>0.248075</td>\n",
" <td>64.601974</td>\n",
" <td>1645.86389</td>\n",
" <td>2025-12-18 19:39:04</td>\n",
" </tr>\n",
" <tr>\n",
" <th>8</th>\n",
" <td>90.001220</td>\n",
" <td>230.000000</td>\n",
" <td>100.442688</td>\n",
" <td>270.836426</td>\n",
" <td>0.083623</td>\n",
" <td>2.748308</td>\n",
" <td>403.442100</td>\n",
" <td>-0.939606</td>\n",
" <td>396.222473</td>\n",
" <td>-16.390797</td>\n",
" <td>...</td>\n",
" <td>239.364624</td>\n",
" <td>0.001362</td>\n",
" <td>4.637836</td>\n",
" <td>325.616100</td>\n",
" <td>909.961060</td>\n",
" <td>-3.605707</td>\n",
" <td>0.248075</td>\n",
" <td>62.869644</td>\n",
" <td>1646.51428</td>\n",
" <td>2025-12-18 19:40:58</td>\n",
" </tr>\n",
" <tr>\n",
" <th>9</th>\n",
" <td>90.001220</td>\n",
" <td>227.700012</td>\n",
" <td>100.442688</td>\n",
" <td>270.715637</td>\n",
" <td>0.089754</td>\n",
" <td>2.644230</td>\n",
" <td>425.803000</td>\n",
" <td>-1.007370</td>\n",
" <td>396.384521</td>\n",
" <td>-16.725641</td>\n",
" <td>...</td>\n",
" <td>239.040222</td>\n",
" <td>0.001361</td>\n",
" <td>5.083293</td>\n",
" <td>349.286682</td>\n",
" <td>949.923500</td>\n",
" <td>-3.702983</td>\n",
" <td>0.208413</td>\n",
" <td>65.850296</td>\n",
" <td>1647.25537</td>\n",
" <td>2025-12-18 20:12:23</td>\n",
" </tr>\n",
" <tr>\n",
" <th>10</th>\n",
" <td>90.001220</td>\n",
" <td>227.700012</td>\n",
" <td>100.442688</td>\n",
" <td>270.661438</td>\n",
" <td>0.078859</td>\n",
" <td>2.628080</td>\n",
" <td>435.462860</td>\n",
" <td>-0.962243</td>\n",
" <td>396.384521</td>\n",
" <td>-17.031452</td>\n",
" <td>...</td>\n",
" <td>239.364441</td>\n",
" <td>0.108690</td>\n",
" <td>5.151842</td>\n",
" <td>353.761353</td>\n",
" <td>949.961060</td>\n",
" <td>-3.733788</td>\n",
" <td>0.207413</td>\n",
" <td>64.788740</td>\n",
" <td>1644.68042</td>\n",
" <td>2025-12-19 03:26:44</td>\n",
" </tr>\n",
" <tr>\n",
" <th>11</th>\n",
" <td>90.001220</td>\n",
" <td>227.700012</td>\n",
" <td>100.442688</td>\n",
" <td>270.725159</td>\n",
" <td>0.078859</td>\n",
" <td>2.628080</td>\n",
" <td>424.227722</td>\n",
" <td>-0.827545</td>\n",
" <td>396.384521</td>\n",
" <td>-17.031452</td>\n",
" <td>...</td>\n",
" <td>239.364441</td>\n",
" <td>0.108690</td>\n",
" <td>4.830516</td>\n",
" <td>353.761353</td>\n",
" <td>893.074000</td>\n",
" <td>-3.733788</td>\n",
" <td>0.207413</td>\n",
" <td>64.788740</td>\n",
" <td>1644.68042</td>\n",
" <td>2025-12-19 03:32:37</td>\n",
" </tr>\n",
" <tr>\n",
" <th>12</th>\n",
" <td>90.001220</td>\n",
" <td>227.700012</td>\n",
" <td>100.442688</td>\n",
" <td>269.977020</td>\n",
" <td>0.078859</td>\n",
" <td>3.799854</td>\n",
" <td>350.998047</td>\n",
" <td>-0.901527</td>\n",
" <td>395.614441</td>\n",
" <td>-18.777557</td>\n",
" <td>...</td>\n",
" <td>226.556274</td>\n",
" <td>0.108690</td>\n",
" <td>5.567447</td>\n",
" <td>312.856700</td>\n",
" <td>887.664734</td>\n",
" <td>-3.445201</td>\n",
" <td>0.188967</td>\n",
" <td>65.376144</td>\n",
" <td>1644.68042</td>\n",
" <td>2025-12-19 03:59:20</td>\n",
" </tr>\n",
" <tr>\n",
" <th>13</th>\n",
" <td>90.001220</td>\n",
" <td>227.700012</td>\n",
" <td>100.442688</td>\n",
" <td>269.098000</td>\n",
" <td>0.078859</td>\n",
" <td>3.325234</td>\n",
" <td>401.663940</td>\n",
" <td>-0.960138</td>\n",
" <td>392.380127</td>\n",
" <td>-16.966602</td>\n",
" <td>...</td>\n",
" <td>235.862518</td>\n",
" <td>0.025195</td>\n",
" <td>2.443661</td>\n",
" <td>305.058400</td>\n",
" <td>963.424100</td>\n",
" <td>-3.654345</td>\n",
" <td>0.188967</td>\n",
" <td>64.824660</td>\n",
" <td>1392.28394</td>\n",
" <td>2025-12-19 04:07:07</td>\n",
" </tr>\n",
" <tr>\n",
" <th>14</th>\n",
" <td>89.000120</td>\n",
" <td>133.000000</td>\n",
" <td>100.442688</td>\n",
" <td>280.595900</td>\n",
" <td>0.078859</td>\n",
" <td>2.347593</td>\n",
" <td>453.182526</td>\n",
" <td>-0.409587</td>\n",
" <td>389.169952</td>\n",
" <td>-15.958172</td>\n",
" <td>...</td>\n",
" <td>238.099854</td>\n",
" <td>0.025195</td>\n",
" <td>5.168004</td>\n",
" <td>417.250244</td>\n",
" <td>852.529200</td>\n",
" <td>-3.702983</td>\n",
" <td>0.268095</td>\n",
" <td>62.195198</td>\n",
" <td>1645.46326</td>\n",
" <td>2025-12-19 04:27:35</td>\n",
" </tr>\n",
" <tr>\n",
" <th>15</th>\n",
" <td>91.002320</td>\n",
" <td>130.000000</td>\n",
" <td>100.442688</td>\n",
" <td>283.037842</td>\n",
" <td>0.109597</td>\n",
" <td>2.547736</td>\n",
" <td>364.069400</td>\n",
" <td>-0.737562</td>\n",
" <td>394.961900</td>\n",
" <td>-18.296043</td>\n",
" <td>...</td>\n",
" <td>226.230900</td>\n",
" <td>0.073010</td>\n",
" <td>4.334429</td>\n",
" <td>338.391663</td>\n",
" <td>793.501953</td>\n",
" <td>-4.727626</td>\n",
" <td>0.003959</td>\n",
" <td>72.804596</td>\n",
" <td>1718.75586</td>\n",
" <td>2025-12-21 14:12:47</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"<p>16 rows × 33 columns</p>\n",
"</div>"
],
"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
}

View File

@@ -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()

View File

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

View File

View File

@@ -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

View File

@@ -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()

View File

@@ -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

View File

@@ -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()

View File

@@ -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"
@@ -261,3 +271,7 @@ ssh:
# --namespace sientia \
# --from-file=ssh-privatekey=git_key \
# --type=kubernetes.io/ssh-auth
# kubectl create secret generic pi-web-api-auth-token \
# --namespace sientia \
# --from-literal=token=your-token-here