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 scouter.utils.clients.pi_web_api_client import PIWebAPIClient from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ 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) 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')