sientia-dataops-scouter_temporal/scouter/activities/api.py

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.to_string()}', metadata=metadata)
        self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata)

        return latest_values.to_dict(orient='records')