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.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']
        end_time = input_data.get('end_time', '*')
        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,
                end_time=end_time,
                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)

        # self.debug(f'Latest values: {latest_values.to_string()}', metadata=metadata)

        # # Normalize the package timestamp
        # valid_timestamp_values = latest_values['timestamp'].dropna()
        # latest_values['timestamp'] = valid_timestamp_values.max()

        self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata)

        return latest_values.to_dict(orient='records')