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.repository.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. This method performs cleanup operations: - Closes the PI Web API client connection - Shuts down SientiaMonitoring services (metrics, notifications) """ self.pi_web_api_client.close() SientiaMonitoring.shutdown(self) @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. The timestamps are normalized to ensure consistency across all records in the response. After converting timestamps to string format, all timestamps are set to the maximum timestamp value (lexicographically) found in the dataset. This ensures all records in a single batch share the same timestamp value. Args: input_data (dict[str, Any]): Activity input parameters. Required fields: - 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: Normalized timestamp string (all records share the same value) - 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, request_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.to_string()}', metadata=metadata) self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata) return latest_values.to_dict(orient='records')