Code import - branch 1.3.0

scouter/activities/activities.py

@@ -0,0 +1,132 @@
+from sientia_do.observability.metrics_controller import MetricsController
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from os import getenv
+    from typing import Any
+
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.observability.logger import Logger
+    from sientia_do.temporal.activities.postgres_sync 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, API):
+    """
+    Unified activities class that combines multiple data processing services.
+
+    This class provides a comprehensive interface for all data processing activities
+    by inheriting from specialized service classes. It handles:
+    - PostgreSQL operations for data persistence
+    - 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
+    interface while maintaining separation of concerns across different data services.
+    """
+
+    def __init__(
+        self,
+        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,
+    ):
+        """
+        Initialize the Activities class with all required services.
+
+        Args:
+            postgres_config (dict[str, Any]): PostgreSQL connection configuration.
+                Required fields: host, port, user, password, dbname, min_connections, max_connections
+            redis_config (dict[str, Any]): Redis connection configuration.
+                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
+        """
+        metrics_controller = MetricsController(
+            logger=logger,
+        )
+
+        # Initialize Postgres
+        Postgres.__init__(
+            self,
+            host=postgres_config['host'],
+            port=postgres_config['port'],
+            user=postgres_config['user'],
+            password=postgres_config['password'],
+            dbname=postgres_config['dbname'],
+            min_connections=postgres_config['min_connections'],
+            max_connections=postgres_config['max_connections'],
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        # Initialize Redis
+        Redis.__init__(
+            self,
+            host=redis_config['host'],
+            port=redis_config['port'],
+            logger=logger,
+            notification_handler=notification_handler,
+            username=redis_config['username'],
+            password=redis_config['password'],
+            metrics_controller=metrics_controller,
+        )
+
+        # Initialize Gates
+        Gates.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        # Initialize MongoDB
+        MongoDB.__init__(
+            self,
+            connection_string=mongodb_config['connection_string'],
+            database_name=mongodb_config['database_name'],
+            logger=logger,
+            notification_handler=notification_handler,
+            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):
+        """
+        Gracefully shutdown all service connections.
+
+        This method ensures proper cleanup of database connections and resources
+        to prevent connection leaks and ensure graceful application termination.
+        """
+        Postgres.close(self)
+        MongoDB.close(self)
+        Redis.close(self)
+        Gates.close(self)
+        API.close(self)

scouter/activities/api.py

@@ -0,0 +1,157 @@
+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_sync 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,
+            headers_config={
+                'Content-Type': 'application/json',
+                'Accept': 'application/json',
+                'x-requested-with': 'piwebapistreams',
+                'User-Agent': 'Aig-Scouter-Agent/1.0',
+            },
+        )
+
+    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')
+    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 = 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:
+            self.send_notification(
+                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')

scouter/activities/gates.py

@@ -0,0 +1,319 @@
+from collections.abc import Hashable
+
+from sientia_do.observability.metrics_controller import MetricsController
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import traceback
+    from typing import Any
+
+    from pandas import DataFrame
+    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.sientia_monitoring import SientiaMonitoring
+
+    from scouter import metrics
+    from scouter.utils.quality.filters import null_values_filter, out_of_bounds_filter
+
+quality_gate_filters = {
+    'NULL_VALUES_FILTER': null_values_filter,
+    'OUT_OF_BOUNDS_FILTER': out_of_bounds_filter,
+}
+
+
+class Gates(SientiaMonitoring):
+    """
+    Data quality gates and filtering operations.
+
+    This class implements data quality validation and filtering for industrial
+    time-series data. It provides:
+    - Configurable data quality filters
+    - Data aggregation functions for time-series data
+    - Comprehensive error handling and notification
+    - Metrics collection for quality monitoring
+
+    The class supports multiple aggregation strategies and quality filters to
+    ensure data integrity and enable flexible data processing workflows.
+    """
+
+    def __init__(
+        self,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        """
+        Initialize the Gates class with logging and notification services.
+
+        Args:
+            logger (Logger): Logger instance for operation logging
+            notification_handler (NotificationHandler): Handler for system notifications
+            metrics_controller (MetricsController): Metrics controller instance
+        """
+        SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
+
+    def close(self):
+        """
+        Close the Gates class.
+        """
+        SientiaMonitoring.shutdown(self)
+
+    def apply_aggregation(
+        self, values: DataFrame, aggr_function: str, metadata: dict[str, Any]
+    ) -> float | None | str:
+        """
+        Apply aggregation function to a group of time-series data.
+
+        This method applies the specified aggregation function to a group of
+        data points. It handles edge cases and provides comprehensive error
+        reporting for invalid aggregation functions.
+
+        Args:
+            values (DataFrame): Group of data points to aggregate (pre-sorted by timestamp)
+            aggr_function (str): Aggregation function to apply.
+                Supported functions: 'lts' (latest), 'avg' (average), 'mdn' (median),
+                'max' (maximum), 'min' (minimum)
+            metadata (dict[str, Any]): Workflow metadata for error reporting
+
+        Returns:
+            float | None | str: Aggregated value, None if no valid data, or 'continue' for errors
+
+        Raises:
+            NotificationError: If invalid aggregation function is specified
+        """
+        aggregation_map = {
+            'lts': lambda x: x.iloc[-1],
+            'avg': lambda x: x.mean(),
+            'mdn': lambda x: x.median(),
+            'max': lambda x: x.max(),
+            'min': lambda x: x.min(),
+        }
+
+        if aggr_function not in aggregation_map:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='AGGREGATION_ISSUES',
+                message=f'Invalid aggregation function: {aggr_function}',
+                block='aggregate_data',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            return 'continue'
+
+        if len(values) == 1:
+            return values['value'].iloc[0]
+
+        if aggr_function == 'lts':
+            return aggregation_map['lts'](values['value'])
+
+        clean_values = values['value'].dropna()
+
+        if clean_values.empty:
+            return None
+
+        return aggregation_map[aggr_function](clean_values)
+
+    @activity.defn(name='aggregate_data')
+    def aggregate_data(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
+        """
+        Aggregate time-series data by tag and name using specified functions.
+
+        This activity processes time-series data by grouping it by tag and name,
+        then applying the configured aggregation functions. It handles data
+        validation and provides comprehensive error reporting.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - data (dict[str, Any]): Time-series data to aggregate
+                - model_tags (dict[str, Any]): Tag configuration with aggregation functions
+
+        Returns:
+            dict[Hashable, Any]: Aggregated data organized by tag and name
+
+        Raises:
+            Exception: If aggregation operation fails
+        """
+
+        metadata = input_data['metadata']
+
+        try:
+            # Convert input data to DataFrame
+            df = DataFrame(input_data['data'])
+
+            self.info(f'Aggregating time series data for {len(df)} rows', metadata=metadata)
+
+            # Sort once by timestamp for all data (more efficient than sorting each group)
+            df = df.sort_values(['tag', 'name', 'timestamp'])
+
+            # Group by tag and name
+            # sort=False since we already sorted
+            grouped = df.groupby(['tag', 'name'], sort=False)
+
+            # Prepare aggregation functions mapping
+            model_tags = input_data['model_tags']
+
+            # Process groups efficiently
+            results = []
+            for (tag, name), group in grouped:
+                # Get the aggregation function from model_tags
+                aggr_function = model_tags.get(name, {}).get('aggr_func', 'lts')
+
+                # Get the latest timestamp (last row since data is sorted)
+                latest_timestamp = group['timestamp'].iloc[-1]
+
+                aggr_value = self.apply_aggregation(group, aggr_function, metadata)
+
+                if aggr_value == 'continue':
+                    continue
+
+                # Store the result directly in list for better performance
+                results.append(
+                    {
+                        'tag': tag,
+                        'name': name,
+                        'value': aggr_value,
+                        'timestamp': latest_timestamp,
+                        'aggregation_function': aggr_function,
+                    }
+                )
+
+            self.info(f'Aggregated data has {len(results)} rows', metadata=metadata)
+
+            # Convert to DataFrame only once at the end if we have results
+            if results:
+                result_df = DataFrame(results)
+
+                self.debug(f'Final aggregated data:\n{result_df.to_string()}', metadata=metadata)
+
+                return result_df.to_dict()
+            else:
+                # Return empty DataFrame dict structure
+                return DataFrame().to_dict()
+
+        except Exception as e:
+            trace = traceback.format_exc()
+
+            self.send_notification(
+                metadata=metadata,
+                notification_id='AGGREGATION_ISSUES',
+                message=f'Error aggregating data: {e}',
+                block='aggregate_data',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+
+            self.error(trace, metadata=metadata)
+            raise e
+
+    @activity.defn(name='data_quality_gate')
+    def data_quality_gate(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
+        """
+        Apply data quality filters to incoming data.
+
+        This activity applies configurable quality filters to validate incoming
+        data. It supports multiple filter types and provides comprehensive
+        error reporting for quality issues.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - data (dict[str, Any]): Data to validate
+                - filters (dict[str, str]): Filter configuration
+                - model_tags (dict[str, Any]): Tag-specific validation rules
+
+        Returns:
+            dict[Hashable, Any]: Filtered data that passes quality validation
+
+        Raises:
+            Exception: If quality validation fails
+        """
+
+        metadata = input_data['metadata']
+
+        filters = input_data['filters']
+        data = DataFrame(input_data['data'])
+        model_tags = input_data['model_tags']
+
+        self.info(f'Applying quality gate to data to {len(data)} rows', metadata=metadata)
+
+        tags = list(model_tags.keys())
+
+        data = data[data['name'].isin(tags)]
+
+        for filter_name, config in filters.items():
+            policy = config['policy']
+            if filter_name not in quality_gate_filters:
+                self.warning(f'Filter {filter_name} not found', metadata=metadata)
+                continue
+
+            try:
+                filtered_data = quality_gate_filters[filter_name](data, model_tags)
+
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id='DATA_QUALITY_GATE_ISSUES',
+                    message=f'Error applying filter {filter_name}: {e}',
+                    block='data_quality_gate',
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace,
+                )
+
+                self.error(trace, metadata=metadata)
+
+            else:
+                if filtered_data.empty:
+                    continue
+
+                message = f'{len(filtered_data)} rows has quality issues: {filter_name}: {policy}'
+                attachment = filtered_data.to_string()
+
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id=f'DATA_QUALITY_GATE_ISSUES__{filter_name}',
+                    message=message,
+                    block='data_quality_gate',
+                    level=NotificationLevel.WARNING,
+                    attachment_content=attachment,
+                )
+
+                if policy == 'DISCARD':
+                    data = data[~data.index.isin(filtered_data.index)]
+
+        self.info(f'Data quality gate applied, final data has {len(data)} rows', metadata=metadata)
+
+        return data.to_dict()
+
+    @activity.defn(name='write_metrics')
+    def write_metrics(self, input_data: dict[str, Any]) -> None:
+        """
+        Write metrics to the database.
+        input_data:
+            metadata: dict[str, Any]
+        """
+        metadata = input_data['metadata']
+        tag_values = DataFrame(input_data['tag_values'])
+
+        self.info(f'Writing metrics for {metadata["model_name"]}', metadata=metadata)
+
+        metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            workflow_name=metadata['workflow_name'],
+        ).inc()
+
+        # Register metrics
+        for _, row in tag_values.iterrows():
+            value = row['value']
+            if value is not None:
+                metrics.TAG_CHANGES_MONITOR.labels(
+                    pod_id=self.pod_id,
+                    model_name=metadata['model_name'],
+                    workflow_name=metadata['workflow_name'],
+                    tag_name=row['variable'],
+                ).set(row['value'])
+
+        self.info(f'Metrics written for {metadata["model_name"]}', metadata=metadata)

scouter/activities/mongodb.py

@@ -0,0 +1,154 @@
+from datetime import UTC
+
+from sientia_do.observability.metrics_controller import MetricsController
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import traceback
+    from datetime import datetime
+    from typing import Any
+
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability.logger import Logger
+    from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+    from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
+    from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
+
+
+class MongoDB(SientiaMonitoring):
+    """
+    MongoDB operations for data retrieval and storage.
+
+    This class provides MongoDB connectivity and operations for the Scouter system.
+    It handles:
+    - Connection management with automatic reconnection
+    - Data retrieval with timestamp-based filtering
+    - Document cleaning and preprocessing
+    - Error handling and notification integration
+
+    The class implements Temporal activities for MongoDB operations, enabling
+    distributed data processing with fault tolerance and monitoring.
+    """
+
+    def __init__(
+        self,
+        connection_string: str,
+        database_name: str,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        """
+        Initialize MongoDB connection and services.
+
+        Args:
+            connection_string (str): MongoDB connection URI string
+            database_name (str): Name of the target database
+            logger (Logger): Logger instance for operation logging
+            notification_handler (NotificationHandler): Handler for system notifications
+
+        Raises:
+            ConnectionError: If MongoDB connection fails
+        """
+
+        self.mongodb_repository = MongoDBRepository(
+            connection_string=connection_string,
+            database_name=database_name,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def close(self):
+        """
+        Close the MongoDB connection.
+        """
+        self.mongodb_repository.close()
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        """
+        Destructor to ensure MongoDB client is closed.
+
+        This destructor ensures that MongoDB connections are properly closed
+        when the object is garbage collected, preventing resource leaks.
+        """
+        self.close()
+
+    @activity.defn(name='load_latest_data')
+    def load_latest_data(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Load the latest data from MongoDB collection since a specified timestamp.
+
+        This activity retrieves data from a MongoDB collection, optionally
+        filtering by timestamp to enable incremental data processing. It
+        handles connection management and provides comprehensive error reporting.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - collection_name (str): Name of the MongoDB collection
+                - last_data_timestamp (str | None): Last processed timestamp for filtering
+
+        Returns:
+            dict[str, Any]: Retrieved data, or empty dict if no data found
+
+        Raises:
+            Exception: If MongoDB operation fails
+        """
+        metadata = input_data['metadata']
+        collection_name = input_data['collection_name']
+        last_data_timestamp = input_data['last_data_timestamp']
+
+        self.info(f'Loading data from MongoDB: {input_data}', metadata=metadata)
+
+        try:
+            if last_data_timestamp is None:
+                data_filter = {}
+            else:
+                data_filter = {
+                    'inserted_at': {
+                        '$gt': datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ)
+                    }
+                }
+
+            self.debug(f'Data filter: {data_filter}', metadata=metadata)
+
+            data = self.mongodb_repository.find(
+                collection_name=collection_name,
+                filters=data_filter,
+                metadata=metadata,
+            )
+
+            self.debug(f'Collected: {data}', metadata=metadata)
+
+            for item in data:
+                item['inserted_at'] = (
+                    item['inserted_at'].replace(tzinfo=UTC).strftime(DATETIME_FORMAT_MS_WITH_TZ)
+                )
+
+            self.info(f'Loaded {len(data)} documents from MongoDB', metadata=metadata)
+
+            self.debug(f'Loaded data: {data}', metadata=metadata)
+
+            return data
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='MONGO_LOAD_ERROR',
+                message=f'Error loading data from MongoDB: {e}',
+                block='load_latest_data',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            raise e

scouter/activities/redis.py

@@ -0,0 +1,311 @@
+from sientia_do.observability.metrics_controller import MetricsController
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import traceback
+    from collections.abc import Hashable
+    from typing import Any
+
+    from pandas import DataFrame
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability.logger import Logger
+    from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+    from sientia_do.repository.redis_repository_sync import RedisRepository
+    from sientia_do.temporal.constants import DATETIME_FORMAT, now
+
+
+class Redis(SientiaMonitoring):
+    """
+    Redis operations for data caching and temporary storage.
+
+    This class extends the base Redis functionality to provide specialized
+    operations for the Scouter system, including:
+    - Data timestamp management for incremental processing
+    - Temporary data storage with configurable TTL
+    - Data grouping and holding for batch processing
+    - Error handling and notification integration
+
+    The class implements Temporal activities for Redis operations, enabling
+    distributed data processing with fault tolerance and monitoring.
+    """
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        username: str,
+        password: str,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        """
+        Initialize Redis connection and services.
+
+        Args:
+            host (str): Redis server hostname or IP address
+            port (int): Redis server port number
+            username (str): Redis authentication username
+            password (str): Redis authentication password
+            logger (Logger): Logger instance for operation logging
+            notification_handler (NotificationHandler): Handler for system notifications
+        """
+        SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
+        self.redis_repository = RedisRepository(
+            host=host,
+            port=port,
+            username=username,
+            password=password,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def close(self):
+        """
+        Close the Redis connection.
+        """
+        self.redis_repository.close()
+        SientiaMonitoring.shutdown(self)
+
+    @activity.defn(name='get_last_data_timestamp')
+    def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
+        """
+        Retrieve the last processed data timestamp from Redis.
+
+        This activity retrieves the timestamp of the last successfully processed
+        data point for a specific workflow and schedule combination. It's used
+        for incremental data processing to avoid reprocessing the same data.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - workflow_name (str): Name of the workflow
+                - schedule_name (str): Name of the data collection schedule
+
+        Returns:
+            str | None: Last processed timestamp string, or None if no previous data exists
+
+        Raises:
+            Exception: If Redis operation fails
+        """
+        metadata = input_data['metadata']
+        key = f'last_data_timestamp:{input_data["workflow_name"]}:{input_data["schedule_name"]}'
+
+        self.info(f'Getting last data timestamp for {key}', metadata=metadata)
+
+        try:
+            data_hold = self.redis_repository.get(key, metadata=metadata)
+        except Exception as e:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_GET_ERROR',
+                message=f'Error getting last data timestamp: {e}',
+                block='get_last_data_timestamp',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            raise e
+
+        self.info(f'Last collected timestamp: {data_hold}', metadata=metadata)
+
+        if not data_hold:
+            return None
+
+        return data_hold
+
+    @activity.defn(name='put_last_data_timestamp')
+    def put_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
+        """
+        Store the last processed data timestamp in Redis.
+
+        This activity stores the timestamp of the most recent data point that
+        has been successfully processed. The timestamp is used for incremental
+        data loading in subsequent workflow executions.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - data (dict[str, Any]): Processed data to extract timestamp from
+                - workflow_name (str): Name of the workflow
+                - schedule_name (str): Name of the data collection schedule
+
+        Returns:
+            str | None: The timestamp that was stored, or None if no data was processed
+
+        Raises:
+            Exception: If Redis operation fails
+        """
+        metadata = input_data['metadata']
+        key = f'last_data_timestamp:{input_data["workflow_name"]}:{input_data["schedule_name"]}'
+
+        self.info(f'Putting last data timestamp for {key}', metadata=metadata)
+
+        data = DataFrame(input_data['data'])
+
+        if data.empty:
+            self.warning('No data to insert', metadata=metadata)
+            return None
+
+        last_data_timestamp = data['inserted_at'].max()
+
+        self.info(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
+
+        try:
+            self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5, metadata=metadata)
+        except Exception as e:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_SET_ERROR',
+                message=f'Error setting last data timestamp: {e}',
+                block='put_last_data_timestamp',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            raise e
+
+        return last_data_timestamp
+
+    @activity.defn(name='group_and_hold_data')
+    def group_and_hold_data(self, input_data: dict[str, Any]) -> dict[Hashable, Any]:
+        """
+        Group data by tags and store temporarily in Redis with TTL.
+
+        This activity organizes processed data by tag names and stores it in Redis
+        with a configurable retention period. The data is grouped to enable
+        efficient batch processing and export operations.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - schedule_name (str): Name of the data collection schedule
+                - workflow_name (str): Name of the workflow
+                - data (dict[str, Any]): Data to group and store
+                - model_id (str): Unique model identifier
+                - model_tags (dict[str, Any]): Tag configuration
+                - retention_time (int): Data retention period in seconds
+
+        Returns:
+            dict[str, Any]: Grouped data organized by tag names
+
+        Raises:
+            Exception: If Redis operation fails
+        """
+        metadata = input_data['metadata']
+
+        self.debug('Grouping and holding data...', metadata=metadata)
+        data = DataFrame(input_data['data'])
+        model_tags = input_data['model_tags']
+        retention_time = input_data['retention_time']
+        fill_missing_tags = input_data['fill_missing_tags']
+
+        key = f'held_data_{input_data["workflow_name"]}_{input_data["schedule_name"]}'
+
+        self.info(f'Getting held data for {key}', metadata=metadata)
+
+        try:
+            data_hold = self.redis_repository.get(key, metadata=metadata)
+        except Exception as e:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_GET_ERROR',
+                message=f'Error getting held data: {e}',
+                block='group_and_hold_data',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            raise e
+
+        if not data_hold:
+            data_hold = {}
+            if data.empty:
+                self.warning('No data to export', metadata=metadata)
+                return data_hold
+
+        self.info(f'Grouping and holding data for {len(data)} rows')
+
+        try:
+            # Remove possibly removed tags
+            tags = list(model_tags.keys())
+            tags.append('timestamp')
+            self.debug(f'Tags to keep: {tags}', metadata=metadata)
+            data_hold = {tag: content for tag, content in data_hold.items() if tag in tags}
+            self.debug(f'Data hold after removing removed tags: {data_hold}', metadata=metadata)
+
+            for _, row in data.iterrows():
+                value = row['value']
+
+                data_hold[row['name']] = value
+
+            if fill_missing_tags:
+                self.debug('Filling missing tags in data package', metadata=metadata)
+                missing_tags = [tag for tag in tags if tag not in list(data_hold.keys())]
+
+                for tag in missing_tags:
+                    data_hold[tag] = None
+
+            data_hold['timestamp'] = (
+                data['timestamp'].max() if not data.empty else data_hold['timestamp']
+            )
+
+            self.redis_repository.set(key, data_hold, ttl=retention_time, metadata=metadata)
+
+            data_hold_df = DataFrame(data_hold, index=[0])
+            data_hold_melted = data_hold_df.melt(
+                id_vars='timestamp', var_name='variable', value_name='value'
+            )
+            data_hold_melted['model_id'] = input_data['model_id']
+
+            data_hold_melted.reset_index(drop=True, inplace=True)
+        except Exception as e:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_SET_ERROR',
+                message=f'Error setting held data: {e}',
+                block='group_and_hold_data',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            raise e
+
+        self.info(f'Data held and melted has {len(data_hold_melted)} rows')
+
+        self.debug(f'Data held and melted:\n {data_hold_melted.to_string()}', metadata=metadata)
+
+        return data_hold_melted.to_dict()
+
+    @activity.defn(name='store_data_package')
+    def store_data_package(self, input_data: dict[str, Any]):
+        """
+        Stores the data package in redis. It's a debug feature and must be toggled on.
+        input_data:
+            metadata: The metadata of the workflow.
+            workflow_name: The name of the workflow.
+            schedule_name: The name of the schedule.
+            held_data: The final scouter output.
+            data: The data used to collect the data.
+        """
+        metadata = input_data['metadata']
+        key = f'data_package_{input_data["workflow_name"]}_{input_data["schedule_name"]}_{now().strftime(DATETIME_FORMAT)}'
+
+        data = DataFrame(input_data['data'])
+        held_data = DataFrame(input_data['held_data'])
+
+        cache = {'data': data.to_dict(), 'held_data': held_data.to_dict()}
+
+        try:
+            self.redis_repository.set(key, cache, ttl=120, metadata=metadata)
+        except Exception as e:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_SET_ERROR',
+                message=f'Error setting data package: {e}',
+                block='store_data_package',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            raise e

scouter/metrics.py

@@ -0,0 +1,25 @@
+from prometheus_client import Counter, Gauge
+
+# Application health and status metrics
+APP_UP = Gauge(
+    'app_up',
+    'Indicates if the application is running (1) or shutting down (0)',
+    ['pod_id'],
+)
+
+# Core labels for consistent metric labeling
+CORE_LABELS = ['pod_id', 'model_name', 'workflow_name']
+
+# Data processing metrics
+LABORIOUS_DATA_WRITTEN_COUNT = Counter(
+    'scouter_laborious_data_written_count',
+    'Number of writings to the database table laborious_data',
+    CORE_LABELS,
+)
+
+# Tag monitoring metrics
+TAG_CHANGES_MONITOR = Gauge(
+    'scouter_tag_changes_monitor',
+    'Current value change of each tag',
+    [*CORE_LABELS, 'tag_name'],
+)

scouter/utils/connectors_config.py

@@ -0,0 +1,19 @@
+from os import getenv
+from typing import Any
+
+
+def build_kafka_config() -> dict[str, Any]:
+    """
+    Build Kafka configuration from environment variables.
+
+    Returns:
+        dict[str, Any]: Kafka configuration dictionary with keys:
+            - bootstrap_servers: Kafka broker addresses (default: localhost:9092)
+            - polling_time: Consumer polling interval in milliseconds (default: 1000)
+            - group_id: Consumer group identifier (default: scouter-group)
+    """
+    return {
+        'bootstrap_servers': getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),
+        'polling_time': int(getenv('KAFKA_POLLING_TIME', '1000')),
+        'group_id': 'scouter-group',
+    }

scouter/utils/quality/filters.py

@@ -0,0 +1,81 @@
+from typing import Any
+
+import numpy as np
+from pandas import DataFrame
+
+
+def check_data_range(value: float | int | None, val_range: list) -> bool:
+    """
+    Check if a value falls outside the specified range.
+
+    This function validates if a numeric value is within the acceptable range
+    defined by the minimum and maximum bounds. It handles edge cases including
+    None values and NaN values.
+
+    Args:
+        value (float | int | None): The numeric value to validate
+        val_range (list): List containing [min_value, max_value] bounds
+
+    Returns:
+        bool: True if value is outside the range, False if within range
+
+    Note:
+        None and NaN values are considered out of range (return True)
+    """
+    if value is None or np.isnan(value):
+        return True
+
+    bottom = val_range[0]
+    up = val_range[-1]
+
+    return value < bottom or value > up
+
+
+def out_of_bounds_filter(df: DataFrame, model_tags: dict[str, Any]) -> DataFrame:
+    """
+    Filter DataFrame rows where values are outside configured ranges.
+
+    This function applies range validation to each row in the DataFrame based
+    on tag-specific configuration. Rows with values outside the configured
+    ranges are filtered out.
+
+    Args:
+        df (DataFrame): DataFrame containing sensor data with 'name' and 'value' columns
+        model_tags (dict[str, Any]): Tag configuration containing data_range for each tag.
+            If a tag doesn't have data_range, it's considered to have infinite bounds.
+
+    Returns:
+        DataFrame: Filtered DataFrame with out-of-bounds values removed
+
+    Note:
+        Tags without data_range configuration are treated as having infinite bounds
+    """
+    return df[
+        df.apply(
+            lambda x: check_data_range(
+                x['value'], model_tags[x['name']].get('data_range', (-np.inf, np.inf))
+            ),
+            axis=1,
+        )
+    ]
+
+
+def null_values_filter(df: DataFrame, _model_tags: dict[str, Any]) -> DataFrame:
+    """
+    Filter DataFrame rows containing null values.
+
+    This function removes rows where the 'value' column contains null values.
+    It's used for data quality filtering to ensure only complete data records
+    are processed.
+
+    Args:
+        df (DataFrame): DataFrame containing sensor data with 'value' column
+        _model_tags (dict[str, Any]): Tag configuration (unused in this filter)
+
+    Returns:
+        DataFrame: Filtered DataFrame with null values removed
+
+    Note:
+        The _model_tags parameter is included for interface consistency but not used
+    """
+    return df[df['value'].isnull()]

scouter/worker/worker.py

@@ -0,0 +1,187 @@
+from temporalio import client, workflow
+from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
+
+with workflow.unsafe.imports_passed_through():
+    import asyncio
+    import os
+    import sys
+
+    from prometheus_client import start_http_server
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.observability.logger import get_logger
+    from sientia_do.temporal.worker.prepare_worker import prepare_worker
+    from sientia_do.utils.connectors_config import (
+        build_api_config,
+        build_mongodb_config,
+        build_postgres_config,
+        build_redis_config,
+    )
+
+    from scouter import metrics
+    from scouter.activities.activities import Activities
+    from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter
+    from scouter.workflow.scouter import Scouter
+    from scouter.workflow.sub_workflows.core_scouter import CoreScouter
+
+# Environment configuration
+POD_ID = os.getenv('HOSTNAME', 'localhost')
+SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091'))
+
+
+async def main():
+    """
+    Main entry point for the Scouter Temporal worker.
+
+    This function initializes and starts all required services:
+    - Prometheus metrics server
+    - Notification handler for MongoDB
+    - Activity implementations for data processing
+    - Temporal client and workers
+    - Multiple task queues for different workflow types
+
+    The worker supports two main task queues:
+    - scouter-queue: Main data processing workflows
+    - fake_data-queue: Test data generation workflows
+
+    Returns:
+        None
+
+    Raises:
+        SystemExit: If worker initialization or execution fails
+    """
+    host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
+
+    logger = get_logger(__name__)
+
+    metadata = {
+        'pod_id': POD_ID,
+        'model_name': '-',
+        'model_id': '-',
+        'workflow_name': '-',
+        'schedule_name': '-',
+    }
+
+    logger.custom_info(f'Starting Worker with pod_id: {POD_ID}', metadata)
+
+    logger.custom_info('Starting prometheus client...', metadata)
+    start_prometheus_server()
+
+    logger.custom_info('Starting Notification Handler...', metadata)
+
+    mongo_config = build_mongodb_config()
+    notification_handler = NotificationHandler(
+        connection_string=mongo_config['connection_string'],
+        database=mongo_config['database_name'],
+        logger=logger,
+        project_name=os.getenv('PROJECT_NAME', 'scouter'),
+    )
+
+    logger.custom_info('Starting Activities...', metadata)
+
+    activities = Activities(
+        logger=logger,
+        notification_handler=notification_handler,
+        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)
+
+    new_runtime = Runtime(
+        telemetry=TelemetryConfig(
+            metrics=PrometheusConfig(bind_address=f'0.0.0.0:{SDK_METRICS_PORT}')
+        )
+    )
+
+    logger.custom_info('Starting Temporal Client...', metadata)
+
+    temporal_client = await client.Client.connect(
+        target_host=host, namespace=os.getenv('TEMPORAL_NAMESPACE', 'scouter'), runtime=new_runtime
+    )
+
+    logger.custom_info('Starting Workers...', metadata)
+
+    workers = [
+        prepare_worker(
+            temporal_client=temporal_client,
+            main_workflow=Scouter,
+            other_workflows=[CoreScouter],
+            activities=[
+                activities.load_latest_data,
+                activities.get_last_data_timestamp,
+                activities.put_last_data_timestamp,
+                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,
+        ),
+        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 = []
+    for w in workers:
+        handlers.append(w.run())
+
+    logger.custom_info('Workers started successfully', metadata)
+
+    try:
+        await asyncio.gather(*handlers)
+
+    except BaseException:  # NOSONAR
+        logger.custom_error('An unhandled exception occurred: %s', metadata=metadata)
+    finally:
+        if notification_handler:
+            notification_handler.shutdown()
+        if activities:
+            activities.shutdown()
+        # Exit with a non-zero status code to indicate failure to Kubernetes
+        metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
+        sys.exit(1)
+
+
+def start_prometheus_server():
+    """
+    Start the Prometheus metrics HTTP server.
+
+    This function initializes the Prometheus metrics server on the configured
+    port and sets the application health status. It's essential for
+    monitoring and observability of the Scouter system.
+
+    Returns:
+        None
+
+    Raises:
+        SystemExit: If metrics server fails to start
+    """
+    try:
+        port = int(os.getenv('HTTP_METRICS_PORT', 9090))
+        start_http_server(port)
+        print(f'Prometheus server started on port {port}.')
+        metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
+    except Exception as e:
+        print(f'Failed to start Prometheus server: {e}')
+        os._exit(1)
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

scouter/workflow/pi_web_api_scouter.py

@@ -0,0 +1,108 @@
+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 (timestamps are normalized)
+        3. Delegates data processing to the CoreScouter workflow
+
+        If no data is retrieved from the PI Web API, the workflow exits early without
+        invoking the CoreScouter workflow.
+
+        Args:
+            input_data (dict[str, Any]): Configuration and parameters for the workflow execution.
+                Required fields:
+                - schedule_name (str): Unique identifier for the data collection schedule
+                - model_name (str): Name of the data model being processed
+                - model_id (str): Unique identifier for the data model
+                - pi_web_api_query (dict[str, Any]): PI Web API query configuration containing:
+                    - endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded')
+                    - period (str): Time period configuration (e.g., '*-1d', '*-1h')
+                    - 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 mapping tag names
+                  to WebIds and processing rules, including:
+                    - webid (str): PI Web API WebId for the tag
+                    - data_range: [min, max] values for data validation
+                    - aggr_func: 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'] = 'pi_web_api_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)

scouter/workflow/scouter.py

@@ -0,0 +1,120 @@
+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='scouter')
+class Scouter:
+    """
+    Main Scouter workflow that orchestrates data ingestion and processing.
+
+    This workflow serves as the entry point for data processing pipelines. It loads
+    data from MongoDB collections, manages data timestamps for incremental processing,
+    and delegates the actual data processing to the CoreScouter workflow.
+
+    The workflow implements a robust data ingestion pattern with:
+    - Incremental data loading based on last processed timestamp
+    - Automatic timestamp management for data continuity
+    - Error handling and retry policies
+    - Child workflow orchestration for data processing
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]) -> None:
+        """
+        Execute the main Scouter workflow.
+
+        This method orchestrates the complete data ingestion process:
+        1. Retrieves the last processed timestamp from Redis
+        2. Loads new data from MongoDB since the last timestamp using collection name
+           format: `raw_{schedule_name}`
+        3. Updates the last processed timestamp with the most recent data point
+        4. Delegates data processing to the CoreScouter workflow
+
+        If no new data is found in MongoDB, the workflow exits early without updating
+        the timestamp or invoking the CoreScouter workflow.
+
+        Args:
+            input_data (dict[str, Any]): Configuration and parameters for the workflow execution.
+                Required fields:
+                - topic (str): The Kafka topic name for data source identification
+                - schedule_name (str): Unique identifier for the data collection schedule
+                - model_name (str): Name of the data model being processed
+                - model_id (str): Unique identifier for the data model
+                - 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
+        """
+
+        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'],
+            }
+        }
+
+        last_data_timestamp = await workflow.execute_local_activity_method(
+            Activities.get_last_data_timestamp,
+            {
+                **metadata,
+                'workflow_name': input_data['workflow_name'],
+                'schedule_name': input_data['schedule_name'],
+            },
+            start_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        data = await workflow.execute_local_activity_method(
+            Activities.load_latest_data,
+            {
+                **metadata,
+                'collection_name': f'raw_{input_data["schedule_name"]}',
+                'last_data_timestamp': last_data_timestamp,
+            },
+            start_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        if not data:
+            return
+
+        await workflow.execute_activity_method(
+            Activities.put_last_data_timestamp,
+            {
+                **metadata,
+                'data': data,
+                'workflow_name': input_data['workflow_name'],
+                'schedule_name': input_data['schedule_name'],
+            },
+            start_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        input_data['data'] = data
+        input_data['metadata'] = metadata
+
+        await workflow.execute_child_workflow('subworkflow.core_scouter', input_data)

scouter/workflow/sub_workflows/core_scouter.py

@@ -0,0 +1,154 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
+    from sientia_do.temporal.policies import retry_policy
+
+    from scouter.activities.activities import Activities
+
+
+@workflow.defn(name='subworkflow.core_scouter')
+class CoreScouter:
+    """
+    Core data processing workflow that handles data quality, aggregation, and export.
+
+    This workflow implements the core data processing pipeline for industrial data:
+    - Data quality validation and filtering
+    - Time-series data aggregation using configurable functions
+    - Data grouping and temporary storage in Redis
+    - Asynchronous export to PostgreSQL for persistent storage
+    - Metrics collection and monitoring
+
+    The workflow is designed for high-throughput data processing with configurable
+    quality gates and aggregation strategies. It is typically invoked as a child
+    workflow by parent workflows such as Scouter or PIWebAPIScouter.
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]) -> None:
+        """
+        Execute the core data processing workflow.
+
+        This method processes industrial time-series data through a series of stages:
+        1. Data Quality Gate: Applies configurable filters for data validation
+        2. Data Aggregation: Groups and aggregates data using specified functions
+        3. Data Grouping: Organizes data by tags and applies retention policies
+        4. Data Export: Persists processed data to PostgreSQL with timestamp conversion
+        5. Metrics Collection: Records processing metrics for monitoring
+
+        The workflow implements early exit conditions:
+        - If held_data is empty after grouping, the workflow exits without exporting
+        - If data export results in zero or negative affected_rows, the workflow exits
+          without writing metrics or storing debug packages
+
+        Args:
+            input_data (dict[str, Any]): Complete workflow configuration and data.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - workflow_name (str): Name of the parent workflow
+                - schedule_name (str): Data collection schedule identifier
+                - model_name (str): Data model name
+                - model_id (str): Unique model identifier
+                - data (dict[str, Any]): Raw time-series data to process
+                - trigger_laborious (bool): Enable intensive processing mode
+                - filters (dict[str, str]): Data quality filter configurations
+                - schema (str): Target database schema
+                - table_name (str): Target database table
+                - retention_time (int): Redis data retention period (seconds)
+                - model_tags (dict[str, Any]): Tag-specific processing rules
+                - fill_missing_tags (bool): Enable filling of missing tag values
+                - debug_data_package (bool, optional): Store data packages for debugging.
+                  When True, stores both raw and processed data in MongoDB for debugging
+
+        Returns:
+            None: This workflow processes data but doesn't return results
+
+        Raises:
+            WorkflowExecutionError: If workflow execution fails
+            ActivityExecutionError: If any activity fails after retry attempts
+        """
+
+        metadata = input_data['metadata']
+
+        filtered_data = await workflow.execute_local_activity_method(
+            Activities.data_quality_gate,
+            {
+                **metadata,
+                'filters': input_data['filters'],
+                'data': input_data['data'],
+                'model_tags': input_data['model_tags'],
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        grouped_data = await workflow.execute_local_activity_method(
+            Activities.aggregate_data,
+            {**metadata, 'data': filtered_data, 'model_tags': input_data['model_tags']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        held_data = await workflow.execute_local_activity_method(
+            Activities.group_and_hold_data,
+            {
+                **metadata,
+                'schedule_name': input_data['schedule_name'],
+                'workflow_name': input_data['workflow_name'],
+                'data': grouped_data,
+                'model_id': input_data['model_id'],
+                'model_tags': input_data['model_tags'],
+                'retention_time': input_data['retention_time'],
+                'fill_missing_tags': input_data['fill_missing_tags'],
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        if held_data == {}:
+            return
+
+        data_exported = await workflow.execute_activity_method(
+            Activities.export_data_to_postgres,
+            {
+                **metadata,
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name'],
+                'data': held_data,
+                'on_conflict': 'ignore',
+                'unique_columns': ['model_id', 'timestamp', 'variable'],
+                'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ},
+            },
+            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,
+            {
+                **metadata,
+                'tag_values': held_data,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        if input_data.get('debug_data_package', False):
+            await workflow.execute_activity_method(
+                Activities.store_data_package,
+                {
+                    **metadata,
+                    'data': input_data['data'],
+                    'held_data': held_data,
+                    'workflow_name': input_data['workflow_name'],
+                    'schedule_name': input_data['schedule_name'],
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )