Merge pull request #31 from Aignosi/feature/SIENTIAPDE-1445

Refactor and Enhance Scouter Workflow Processing, Configuration, and Robustness

orchestrator/activities/formatters.py

@@ -2,10 +2,10 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     import json
-    from collections.abc import Hashable
+    from collections.abc import Callable, Hashable
     from logging import Logger
     from math import ceil
-    from typing import Any
+    from typing import Any, TypedDict
 
     from pandas import DataFrame
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -18,6 +18,7 @@ with workflow.unsafe.imports_passed_through():
         drift,
         gather_read_tags,
         minimal_retrain,
+        pi_web_api_scouter,
         predictions_batch,
         scouter,
         simple_metrics,
@@ -26,6 +27,68 @@ with workflow.unsafe.imports_passed_through():
 topic_separator = '\n ========== \n'
 
 
+class ScheduleType(TypedDict):
+    """
+    Type definition for schedule type configuration entries.
+
+    Each schedule type entry maps a workflow type string to its corresponding
+    namespace and configuration builder function.
+
+    Attributes:
+        namespace (str): The Temporal namespace where workflows of this type execute.
+            Valid values: 'scouter', 'laborious'
+        function (Callable): Configuration builder function that transforms pipeline
+            configuration into Temporal-compatible workflow arguments
+    """
+
+    namespace: str
+    function: Callable
+
+
+schedule_types: dict[str, ScheduleType] = {
+    'scouter': {
+        'namespace': 'scouter',
+        'function': scouter,
+    },
+    'pi_web_api_scouter': {
+        'namespace': 'scouter',
+        'function': pi_web_api_scouter,
+    },
+    'predictions_batch': {
+        'namespace': 'laborious',
+        'function': predictions_batch,
+    },
+    'minimal_retrain': {
+        'namespace': 'laborious',
+        'function': minimal_retrain,
+    },
+    'drift': {
+        'namespace': 'laborious',
+        'function': drift,
+    },
+    'simple_metrics': {
+        'namespace': 'laborious',
+        'function': simple_metrics,
+    },
+}
+"""
+Registry mapping workflow types to their namespace and configuration builder functions.
+
+Supported workflow types:
+    - scouter: OPC data collection using OPC UA protocol
+    - pi_web_api_scouter: Data collection using PI Web API
+    - predictions_batch: ML model prediction workflows with OPC write-back
+    - minimal_retrain: Model retraining workflows using SQL queries
+    - drift: Data drift detection and monitoring workflows
+    - simple_metrics: Model performance metrics computation workflows
+
+Each entry specifies:
+    - namespace: Target Temporal namespace for workflow execution
+    - function: Configuration builder that transforms MongoDB pipeline config
+                into Temporal workflow arguments
+"""
+
+
 class Formatters(SientiaMonitoring):
     """
     Schedule and slot configuration formatting and notification filtering activity.
@@ -115,37 +178,18 @@ class Formatters(SientiaMonitoring):
         }
 
         for pipeline in pipelines:
-            if pipeline['workflow_type'] == 'scouter':
+            workflow_type = pipeline['workflow_type']
-                schedule_config[self.scouter_namespace][pipeline['schedule_name']] = {
+
-                    **scouter(pipeline),
+            if workflow_type not in schedule_types:
-                    'updated_at': pipeline.get(
+                self.error(f'Workflow type {workflow_type} not supported', metadata=metadata)
-                        'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
+                continue
-                    ),
+
-                }
+            schedule_type = schedule_types[workflow_type]
-            elif pipeline['workflow_type'] == 'predictions_batch':
+            namespace = schedule_type['namespace']
-                schedule_config[self.laborious_namespace][pipeline['schedule_name']] = {
+            function = schedule_type['function']
-                    **predictions_batch(pipeline),
+
-                    'updated_at': pipeline.get(
+            schedule_config[namespace][pipeline['schedule_name']] = {
-                        'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
+                **function(pipeline),
-                    ),
-                }
-            elif pipeline['workflow_type'] == 'minimal_retrain':
-                schedule_config[self.laborious_namespace][pipeline['schedule_name']] = {
-                    **minimal_retrain(pipeline),
-                    'updated_at': pipeline.get(
-                        'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
-                    ),
-                }
-            elif pipeline['workflow_type'] == 'drift':
-                schedule_config[self.laborious_namespace][pipeline['schedule_name']] = {
-                    **drift(pipeline),
-                    'updated_at': pipeline.get(
-                        'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
-                    ),
-                }
-            elif pipeline['workflow_type'] == 'simple_metrics':
-                schedule_config[self.laborious_namespace][pipeline['schedule_name']] = {
-                    **simple_metrics(pipeline),
                 'updated_at': pipeline.get(
                     'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
                 ),

orchestrator/activities/mongo_db.py

@@ -393,7 +393,9 @@ class MongoDB(SientiaMonitoring):
         created_indexes = []
 
         for _pipeline_name, pipeline_config in pipelines.items():
-            collection = pipeline_config['topic']
+            collection = pipeline_config.get('topic', None)
+            if not collection:
+                continue
 
             try:
                 # Check if collection exists

orchestrator/utils/orchestrator_functions.py

@@ -1,5 +1,7 @@
 from typing import Any
 
+from orchestrator.utils.converters import parse_frequency
+
 
 def common_config(config: dict[str, Any]):
     """
@@ -7,15 +9,22 @@ def common_config(config: dict[str, Any]):
 
     Args:
         config (dict[str, Any]): Pipeline configuration containing:
-            - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch')
+            - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch', 'drift')
-            - schedule_name (str): Name of the schedule
+            - schedule_name (str): Unique name identifier for this schedule
-            - frequency (str, optional): Frequency of execution (default: '1m')
+            - model_id (str): MongoDB ID of the associated model
-            - max_retry_policy (int, optional): Maximum retry attempts (default: 1)
+            - model (dict): Model configuration containing:
-            - model_id (str): ID of the model
+                - name (str): Human-readable name of the model
-            - models (dict): Model configuration containing 'name' field
+                - model_config (dict, optional): Additional model-specific configuration
+            - frequency (str, optional): Execution frequency (default: '1m')
+                Format: '{number}{unit}' where unit is 's', 'm', 'h', or 'd'
+            - offset (str, optional): Schedule offset/delay (default: '0m')
+            - max_retry_policy (int, optional): Maximum retry attempts on failure (default: 1)
+            - execution_timeout_seconds (int, optional): Workflow execution timeout (default: 300)
+            - task_timeout_seconds (int, optional): Individual task timeout (default: 300)
 
     Returns:
-        dict[str, Any]: Common configuration dictionary with extracted parameters.
+        dict[str, Any]: Common configuration dictionary with standardized parameters
+            for Temporal workflow execution
     """
     model = config['model']
     return {
@@ -41,13 +50,14 @@ def drift(config: dict[str, Any]):
 
     Args:
         config (dict[str, Any]): Pipeline configuration containing:
-            - interval_minutes (int, optional): Detection interval in minutes (default: 60)
+            - interval_minutes (int, optional): Time window for data comparison in minutes (default: 60)
-            - drift_metrics (list[str], optional): List of drift metrics to compute
+            - drift_metrics (list[str], optional): Statistical metrics to compute (default:
-                (default: ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
+                ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
             - Additional fields from common_config
 
     Returns:
-        dict[str, Any]: Drift configuration with workflow type set to 'drift'
+        dict[str, Any]: Drift detection configuration with source/target tables,
+            time interval, and metrics specifications. Results stored in 'drift_metrics' table
     """
     return {
         **common_config(config),
@@ -76,7 +86,9 @@ def simple_metrics(config: dict[str, Any]):
             - Additional fields from common_config
 
     Returns:
-        dict[str, Any]: Simple metrics configuration with workflow type set to 'simple_metrics'
+        dict[str, Any]: Simple metrics configuration with source tables (predictions and
+            actual data), target table, time interval, and metrics list. Results stored in
+            'simple_metrics' table
     """
     return {
         **common_config(config),
@@ -95,12 +107,15 @@ def minimal_retrain(config: dict[str, Any]):
 
     Args:
         config (dict[str, Any]): Pipeline configuration containing:
-            - schedule_name (str): Name of the schedule
+            - query (str): SQL query to retrieve training data. Should return features
-            - query (str): SQL query for retraining
+                and target variable in expected format
+            - datetime_columns (list[str], optional): Column names to parse as datetime
+                for proper temporal handling (default: [])
             - Additional fields from common_config
 
     Returns:
-        dict[str, Any]: Minimal retrain configuration with workflow type set to 'minimal_retrain'.
+        dict[str, Any]: Minimal retrain configuration with SQL query, database settings,
+            and datetime column specifications. Retraining logs are stored in 'log_retrain' table
     """
     return {
         **common_config(config),
@@ -111,29 +126,63 @@ def minimal_retrain(config: dict[str, Any]):
     }
 
 
+def base_scouter(config: dict[str, Any]):
+    """
+    Build base scouter configuration shared by all scouter workflow types.
+
+    Creates the foundational configuration for OPC data collection workflows,
+    including filter policies, database settings, and data retention parameters.
+    This configuration is extended by specific scouter implementations (OPC UA, PI Web API).
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - filters (list[dict], optional): List of filter configurations with:
+                - filter_name (str): Name of the filter
+                - policy (str): Filter policy to apply
+            - tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60)
+            - debug_data_package (bool, optional): Enable debug data package logging (default: False)
+            - fill_missing_tags (bool, optional): Fill missing tags with interpolation (default: False)
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Base scouter configuration with filters, database settings,
+            and retention policies
+    """
+
+    filters = {}
+    for f in config.get('filters', []):
+        filters[f['filter_name']] = {'policy': f['policy']}
+
+    return {
+        **common_config(config),
+        'trigger_laborious': False,
+        'filters': filters,
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': config.get('tag_retention_minutes', 60) * 60,
+        'debug_data_package': config.get('debug_data_package', False),
+        'fill_missing_tags': config.get('fill_missing_tags', False),
+    }
+
+
 def scouter(config: dict[str, Any]):
     """
     Build scouter configuration from pipeline config.
 
     Args:
         config (dict[str, Any]): Pipeline configuration containing:
-            - filters (list[dict], optional): List of filter configurations
+            - schedule_name (str): Name of the schedule (used for topic generation)
             - read_tags (list[dict]): List of tag configurations with:
-                - filter_name (str): Name of the filter
+                - tag_name (str): Name of the tag to read
-                - policy (str): Filter policy
+                - aggr_func (str, optional): Aggregation function for data collection (default: 'lts')
-                - tag_name (str): Name of the tag
+                    Common values: 'lts' (last), 'avg' (average), 'min', 'max', 'sum'
-                - aggr_func (str, optional): Aggregation function (default: 'lts')
+                - data_range (list[int], optional): Valid data range [min, max] (default: [-100, 100])
-                - data_range (list[int], optional): Data range limits (default: [-100, 100])
+            - Additional fields from base_scouter
-            - tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60)
-            - debug_data_package (bool, optional): Enable debug data package (default: False)
-            - Additional fields from common_config
 
     Returns:
-        dict[str, Any]: Scouter configuration with topic, filters, tags, and retention settings.
+        dict[str, Any]: OPC UA scouter configuration with Kafka topic, tag mappings,
+            filters, and retention settings. Topic name follows pattern: 'raw_{schedule_name}'
     """
-    filters = {}
-    for f in config.get('filters', []):
-        filters[f['filter_name']] = {'policy': f['policy']}
 
     tags = {}
     for tag in config['read_tags']:
@@ -143,16 +192,65 @@ def scouter(config: dict[str, Any]):
         }
 
     return {
-        **common_config(config),
+        **base_scouter(config),
         'topic': f'raw_{config["schedule_name"]}',
-        'trigger_laborious': False,
-        'filters': filters,
-        'schema': 'sientia_data',
-        'table_name': 'laborious_data',
-        'retention_time': config.get('tag_retention_minutes', 60) * 60,
         'model_tags': tags,
-        'debug_data_package': config.get('debug_data_package', False),
+    }
-        'fill_missing_tags': config.get('fill_missing_tags', False),
+
+
+def pi_web_api_scouter(config: dict[str, Any]):
+    """
+    Build PI Web API scouter configuration from pipeline config.
+
+    Creates a data collection workflow configuration for OSIsoft PI servers using
+    the PI Web API REST interface. Configures tag mappings with WebIDs, aggregation
+    functions, and API query parameters including timeout management.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - read_tags (list[dict]): List of tag configurations with:
+                - tag_name (str): Name of the tag
+                - webid (str): PI Web API WebID for the tag
+                - aggr_func (str, optional): Aggregation function (default: 'lts')
+                - data_range (list[int], optional): Valid data range (default: [-100, 100])
+            - pi_web_api_config (dict): PI Web API connection settings with:
+                - endpoint (str): PI Web API endpoint URL
+                - period (str, optional): Time period for data retrieval (default: '*-1d')
+                - max_count (int, optional): Maximum number of values to retrieve (default: 1)
+                - api_timeout (int, optional): API request timeout in seconds
+            - Additional fields from base_scouter
+
+    Returns:
+        dict[str, Any]: PI Web API scouter configuration with tag mappings and query settings.
+            API timeout is automatically adjusted to not exceed workflow frequency.
+    """
+    tags = {}
+    for tag, tag_config in config['read_tags'].items():
+        tags[tag] = {
+            'webid': tag_config['webid'],
+            'aggr_func': tag_config.get('aggr_func', 'lts'),
+            'data_range': tag_config.get('data_range', [-100, 100]),
+        }
+
+    base_config = base_scouter(config)
+
+    pi_web_api_config = config['pi_web_api_config']
+
+    config_timeout = pi_web_api_config.get('api_timeout', None)
+    frequency = parse_frequency(base_config['frequency'])
+
+    if config_timeout is None or config_timeout > frequency:
+        config_timeout = frequency
+
+    return {
+        **base_config,
+        'model_tags': tags,
+        'pi_web_api_query': {
+            'endpoint': pi_web_api_config['endpoint'],
+            'period': pi_web_api_config.get('period', '*-1d'),
+            'max_count': pi_web_api_config.get('max_count', 1),
+            'api_timeout': config_timeout,
+        },
     }
 
 
@@ -181,13 +279,25 @@ def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[
 
 def process_path_priority(path_priority: list[str]):
     """
-    Process and normalize path priority list to ensure it contains the required priorities.
+    Process and normalize path priority list for filter policy execution order.
+
+    Validates and normalizes the priority list used to determine the order in which
+    filter policies are evaluated in prediction workflows. Invalid priorities are
+    removed, missing required priorities are appended, and the list is truncated to
+    exactly 3 elements.
+
+    Valid priorities define workflow behavior when filters are triggered:
+        - STOP: Halt workflow execution immediately
+        - CONTINUE: Proceed to next step despite filter trigger
+        - REPEAT: Retry the current step
 
     Args:
-        path_priority (list[str]): List of path priorities to process.
+        path_priority (list[str]): User-provided list of path priorities. May contain
+            invalid values or be incomplete.
 
     Returns:
-        list[str]: Normalized path priority list with exactly 3 elements: ["STOP", "CONTINUE", "REPEAT"].
+        list[str]: Normalized path priority list with exactly 3 elements in user-specified
+            or default order. Default order when priorities are missing: ["STOP", "CONTINUE", "REPEAT"]
     """
     for priority in path_priority[:]:
         if priority not in ['STOP', 'CONTINUE', 'REPEAT']:
@@ -206,20 +316,25 @@ def predictions_batch(config: dict[str, Any]):
 
     Args:
         config (dict[str, Any]): Pipeline configuration containing:
-            - write_tags (list[dict]): List of tag configurations with:
+            - query (str): SQL query to retrieve input data for predictions
-                - server_id (str): ID of the OPC server
+            - write_tags (list[dict]): List of OPC tag configurations for write-back with:
-                - type (str): Tag type ('prediction' or 'confidence')
+                - server_id (str): ID of the target OPC server
-                - addr (str): Tag address
+                - type (str): Tag type - 'prediction' (model output) or 'confidence' (prediction confidence)
-                - data_type (str, optional): Data type (default: 'float')
+                - addr (str): OPC tag address/path
-            - path_priority (list[str], optional): List of path priorities (default: ["STOP", "CONTINUE", "REPEAT"])
+                - data_type (str, optional): OPC data type (default: 'float')
-            - input_filters (list[dict], optional): List of input filter configurations
+            - datetime_columns (list[str], optional): Column names to parse as datetime (default: [])
-            - mlflow_transform_filters (list[dict], optional): List of MLflow transform filter configurations
+            - path_priority (list[str], optional): Filter policy execution order (default: ["STOP", "CONTINUE", "REPEAT"])
-            - mlflow_predict_filters (list[dict], optional): List of MLflow predict filter configurations
+            - input_filters (list[dict], optional): Input data validation filters
-            - model_retention_minutes (int, optional): Model retention time in minutes (default: 60)
+            - mlflow_transform_filters (list[dict], optional): Transform stage filters
+            - mlflow_predict_filters (list[dict], optional): Prediction stage filters
+            - model_retention_minutes (int, optional): Data retention time in minutes (default: 60)
+            - save_transform (bool, optional): Save transformed data to database (default: True)
+            - predictions_storage_policy (str, optional): Prediction storage policy (default: 'lts:1')
             - Additional fields from common_config
 
     Returns:
-        dict[str, Any]: Predictions batch configuration with OPC output config, filters, and path priority.
+        dict[str, Any]: Complete predictions batch configuration with OPC output mappings,
+            multi-stage filters, SQL query, and retention policies
     """
     tags: dict[str, Any] = {}
     for tag in config.get('write_tags', []):
@@ -290,6 +405,9 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
 
     # Get all read tags from pipelines
     for pipeline in pipelines:
+        if pipeline['workflow_type'] != 'scouter':
+            continue
+
         for tag in pipeline.get('read_tags', []):
             tag_string = f'{tag["server_id"]}:{tag["tag_address"]}'
             if tag_string not in tags:

tests/orchestrator/activities/test_formatters.py

@@ -21,6 +21,9 @@ def formatters():
     formatters.send_notification = MagicMock()
     formatters.send_notification_async = AsyncMock()
     formatters.emit_metric = AsyncMock()
+    formatters.error = MagicMock()
+    formatters.info = MagicMock()
+    formatters.debug = MagicMock()
     return formatters
 
 
@@ -35,31 +38,42 @@ metadata = {
 
 
 @mark.asyncio
-@patch('orchestrator.activities.formatters.scouter', return_value={'test_scouter': 'test_scouter'})
+async def test_process_schedules(formatters):
-@patch(
+    mock_scouter = MagicMock(return_value={'test_scouter': 'test_scouter'})
-    'orchestrator.activities.formatters.predictions_batch',
+    mock_predictions_batch = MagicMock(
-    return_value={'test_predictions_batch': 'test_predictions_batch'},
+        return_value={'test_predictions_batch': 'test_predictions_batch'}
-)
+    )
-@patch(
+    mock_minimal_retrain = MagicMock(return_value={'test_minimal_retrain': 'test_minimal_retrain'})
-    'orchestrator.activities.formatters.minimal_retrain',
+    mock_drift = MagicMock(return_value={'test_drift': 'test_drift'})
-    return_value={'test_minimal_retrain': 'test_minimal_retrain'},
+    mock_simple_metrics = MagicMock(return_value={'test_simple_metrics': 'test_simple_metrics'})
-)
+
-@patch(
+    mock_schedule_types = {
-    'orchestrator.activities.formatters.drift',
+        'scouter': {
-    return_value={'test_drift': 'test_drift'},
+            'namespace': 'scouter',
-)
+            'function': mock_scouter,
-@patch(
+        },
-    'orchestrator.activities.formatters.simple_metrics',
+        'pi_web_api_scouter': {
-    return_value={'test_simple_metrics': 'test_simple_metrics'},
+            'namespace': 'scouter',
-)
+            'function': mock_scouter,
-async def test_process_schedules(
+        },
-    mock_simple_metrics,
+        'predictions_batch': {
-    mock_drift,
+            'namespace': 'laborious',
-    mock_minimal_retrain,
+            'function': mock_predictions_batch,
-    mock_predictions_batch,
+        },
-    mock_scouter,
+        'minimal_retrain': {
-    formatters,
+            'namespace': 'laborious',
-):
+            'function': mock_minimal_retrain,
+        },
+        'drift': {
+            'namespace': 'laborious',
+            'function': mock_drift,
+        },
+        'simple_metrics': {
+            'namespace': 'laborious',
+            'function': mock_simple_metrics,
+        },
+    }
+
     input_data = {
         'pipelines': [
             {
@@ -100,6 +114,7 @@ async def test_process_schedules(
         ]
     }
 
+    with patch('orchestrator.activities.formatters.schedule_types', mock_schedule_types):
         result = await formatters.process_schedules(input_data)
 
     assert result == {
@@ -133,6 +148,79 @@ async def test_process_schedules(
     mock_simple_metrics.assert_called_once_with(input_data['pipelines'][4])
 
 
+@mark.asyncio
+async def test_process_schedules_with_invalid_workflow_type(formatters):
+    """Test that process_schedules handles invalid workflow types correctly"""
+    mock_scouter = MagicMock(return_value={'test_scouter': 'test_scouter'})
+
+    mock_schedule_types = {
+        'scouter': {
+            'namespace': 'scouter',
+            'function': mock_scouter,
+        },
+    }
+
+    pipelines = [
+        {
+            'schedule_name': 'test_schedule_name_valid',
+            'workflow_type': 'scouter',
+            'model_name': 'test_model_name',
+            'model_id': 'test_model_id',
+            'updated_at': '2021-01-01',
+        },
+        {
+            'schedule_name': 'test_schedule_name_invalid',
+            'workflow_type': 'invalid_workflow_type',
+            'model_name': 'test_model_name',
+            'model_id': 'test_model_id',
+            'updated_at': '2021-01-02',
+        },
+        {
+            'schedule_name': 'test_schedule_name_valid2',
+            'workflow_type': 'scouter',
+            'model_name': 'test_model_name',
+            'model_id': 'test_model_id',
+            'updated_at': '2021-01-03',
+        },
+    ]
+
+    input_data = {
+        'pipelines': pipelines,
+        'metadata': {
+            'schedule_name': 'test_schedule',
+            'workflow_name': 'test_workflow',
+        },
+    }
+
+    with patch('orchestrator.activities.formatters.schedule_types', mock_schedule_types):
+        result = await formatters.process_schedules(input_data)
+
+    # Assert that error was called for invalid workflow type
+    formatters.error.assert_called_once_with(
+        'Workflow type invalid_workflow_type not supported', metadata=input_data['metadata']
+    )
+
+    # Assert that only valid pipelines were processed
+    assert result == {
+        'scouter': {
+            'test_schedule_name_valid': {
+                'test_scouter': 'test_scouter',
+                'updated_at': '2021-01-01',
+            },
+            'test_schedule_name_valid2': {
+                'test_scouter': 'test_scouter',
+                'updated_at': '2021-01-03',
+            },
+        },
+        'laborious': {},
+    }
+
+    # Assert that the mock function was called only for valid pipelines
+    assert mock_scouter.call_count == 2
+    mock_scouter.assert_any_call(pipelines[0])
+    mock_scouter.assert_any_call(pipelines[2])
+
+
 @mark.asyncio
 @patch(
     'orchestrator.activities.formatters.gather_read_tags',

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -1,12 +1,14 @@
 from unittest.mock import call, patch
 
 from orchestrator.utils.orchestrator_functions import (
+    base_scouter,
     build_tag_config,
     common_config,
     drift,
     gather_read_tags,
     minimal_retrain,
     overlap_filter_config,
+    pi_web_api_scouter,
     predictions_batch,
     process_path_priority,
     scouter,
@@ -219,7 +221,15 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi
         [call({'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, config['input_filters'])]
     )
     mock_overlap_filter_config.assert_has_calls(
-        [call({'API_ERROR': {'policy': 'STOP', 'config': {}}}, config['mlflow_transform_filters'])]
+        [
+            call(
+                {
+                    'EMPTY_DATA': {'policy': 'STOP', 'config': {}},
+                    'API_ERROR': {'policy': 'STOP', 'config': {}},
+                },
+                config['mlflow_transform_filters'],
+            )
+        ]
     )
     mock_overlap_filter_config.assert_has_calls(
         [call({'API_ERROR': {'policy': 'STOP', 'config': {}}}, config['mlflow_predict_filters'])]
@@ -262,6 +272,7 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi
 def test_gather_read_tags():
     pipelines = [
         {
+            'workflow_type': 'scouter',
             'schedule_name': 'test_schedule',
             'read_tags': [
                 {
@@ -272,6 +283,7 @@ def test_gather_read_tags():
             ],
         },
         {
+            'workflow_type': 'scouter',
             'schedule_name': 'test_schedule2',
             'read_tags': [
                 {
@@ -314,6 +326,103 @@ def test_gather_read_tags():
     assert result == expected
 
 
+def test_gather_read_tags_filters_non_scouter_workflows():
+    """Test that gather_read_tags only processes scouter workflow types and ignores others"""
+    pipelines = [
+        {
+            'workflow_type': 'scouter',
+            'schedule_name': 'test_scouter_schedule',
+            'read_tags': [
+                {
+                    'server_id': '1',
+                    'server_name': 'test_server_name',
+                    'tag_address': 'test_tag_address',
+                }
+            ],
+        },
+        {
+            'workflow_type': 'predictions_batch',
+            'schedule_name': 'test_predictions_schedule',
+            'read_tags': [
+                {
+                    'server_id': '2',
+                    'server_name': 'test_server_name2',
+                    'tag_address': 'test_tag_address_predictions',
+                }
+            ],
+        },
+        {
+            'workflow_type': 'minimal_retrain',
+            'schedule_name': 'test_retrain_schedule',
+            'read_tags': [
+                {
+                    'server_id': '3',
+                    'server_name': 'test_server_name3',
+                    'tag_address': 'test_tag_address_retrain',
+                }
+            ],
+        },
+        {
+            'workflow_type': 'pi_web_api_scouter',
+            'schedule_name': 'test_pi_web_api_schedule',
+            'read_tags': [
+                {
+                    'server_id': '4',
+                    'server_name': 'test_server_name4',
+                    'tag_address': 'test_tag_address_pi_web_api',
+                }
+            ],
+        },
+        {
+            'workflow_type': 'drift',
+            'schedule_name': 'test_drift_schedule',
+            'read_tags': [
+                {
+                    'server_id': '5',
+                    'server_name': 'test_server_name5',
+                    'tag_address': 'test_tag_address_drift',
+                }
+            ],
+        },
+        {
+            'workflow_type': 'scouter',
+            'schedule_name': 'test_scouter_schedule2',
+            'read_tags': [
+                {
+                    'server_id': '1',
+                    'server_name': 'test_server_name',
+                    'tag_address': 'test_tag_address2',
+                }
+            ],
+        },
+    ]
+
+    result = gather_read_tags(pipelines)
+
+    # Only scouter workflow types should be included
+    expected = {
+        '1:test_tag_address': {
+            'server_id': '1',
+            'server_name': 'test_server_name',
+            'tag_address': 'test_tag_address',
+            'topics': ['raw_test_scouter_schedule'],
+        },
+        '1:test_tag_address2': {
+            'server_id': '1',
+            'server_name': 'test_server_name',
+            'tag_address': 'test_tag_address2',
+            'topics': ['raw_test_scouter_schedule2'],
+        },
+    }
+
+    assert result == expected
+    # Ensure non-scouter pipelines are not included
+    assert '2:test_tag_address_predictions' not in result
+    assert '3:test_tag_address_retrain' not in result
+    assert '4:test_tag_address_pi_web_api' not in result
+    assert '5:test_tag_address_drift' not in result
+
+
 def test_build_tag_config():
     tags = [
         {
@@ -403,3 +512,203 @@ def test_build_tag_config():
     }
 
     assert result == (expected, ['3'])
+
+
+def test_base_scouter():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [
+            {'filter_name': 'test_filter_name', 'policy': 'test_policy'},
+            {'filter_name': 'test_filter_name2', 'policy': 'test_policy2'},
+        ],
+        'tag_retention_minutes': 30,
+        'debug_data_package': True,
+        'fill_missing_tags': True,
+    }
+    result = base_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'trigger_laborious': False,
+        'filters': {
+            'test_filter_name': {'policy': 'test_policy'},
+            'test_filter_name2': {'policy': 'test_policy2'},
+        },
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 30 * 60,
+        'debug_data_package': True,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': True,
+    }
+    assert result == expected
+
+
+def test_pi_web_api_scouter():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}],
+        'read_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+                'aggr_func': 'test_aggr_func',
+                'data_range': [1, 2],
+            }
+        },
+        'tag_retention_minutes': 10,
+        'pi_web_api_config': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-2d',
+            'max_count': 5,
+            'api_timeout': 30,
+        },
+        'frequency': '1m',
+    }
+    result = pi_web_api_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'trigger_laborious': False,
+        'filters': {'test_filter_name': {'policy': 'test_policy'}},
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 10 * 60,
+        'model_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+                'aggr_func': 'test_aggr_func',
+                'data_range': [1, 2],
+            }
+        },
+        'debug_data_package': False,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': False,
+        'pi_web_api_query': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-2d',
+            'max_count': 5,
+            'api_timeout': 30,
+        },
+    }
+    assert result == expected
+
+
+def test_pi_web_api_scouter_with_timeout_greater_than_frequency():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [],
+        'read_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+            }
+        },
+        'tag_retention_minutes': 10,
+        'pi_web_api_config': {
+            'endpoint': 'https://test-endpoint.com',
+            'api_timeout': 120,
+        },
+        'frequency': '1m',
+    }
+    result = pi_web_api_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'trigger_laborious': False,
+        'filters': {},
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 10 * 60,
+        'model_tags': {
+            'test_tag_name': {'webid': 'test_webid', 'aggr_func': 'lts', 'data_range': [-100, 100]}
+        },
+        'debug_data_package': False,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': False,
+        'pi_web_api_query': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-1d',
+            'max_count': 1,
+            'api_timeout': 60,
+        },
+    }
+    assert result == expected
+
+
+def test_pi_web_api_scouter_with_no_timeout():
+    config = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'filters': [],
+        'read_tags': {
+            'test_tag_name': {
+                'webid': 'test_webid',
+            }
+        },
+        'tag_retention_minutes': 10,
+        'pi_web_api_config': {
+            'endpoint': 'https://test-endpoint.com',
+        },
+        'frequency': '30s',
+    }
+    result = pi_web_api_scouter(config)
+    expected = {
+        'workflow_type': 'scouter',
+        'schedule_name': 'test_schedule',
+        'frequency': '30s',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'trigger_laborious': False,
+        'filters': {},
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': 10 * 60,
+        'model_tags': {
+            'test_tag_name': {'webid': 'test_webid', 'aggr_func': 'lts', 'data_range': [-100, 100]}
+        },
+        'debug_data_package': False,
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+        'fill_missing_tags': False,
+        'pi_web_api_query': {
+            'endpoint': 'https://test-endpoint.com',
+            'period': '*-1d',
+            'max_count': 1,
+            'api_timeout': 30,
+        },
+    }
+    assert result == expected

values.yaml

@@ -151,7 +151,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
   - name: GITHUB_BRANCH
-    value: "feature/SIENTIAPDE-1273"
+    value: "feature/SIENTIAPDE-1445"
   - name: PYTHON_APP
     value: "orchestrator.worker.worker"
 
@@ -182,17 +182,17 @@ env:
   - name: MONGODB_TTL_INDEX_HOURS
     value: "1"
 
-  - name: EMAIL_SENDER
+  # - name: EMAIL_SENDER
-    value: "vitor.santos@aignosi.com.br"
+  #   value: "vitor.santos@aignosi.com.br"
-  - name: EMAIL_SENDER_PASSWORD
+  # - name: EMAIL_SENDER_PASSWORD
-    valueFrom:
+  #   valueFrom:
-      secretKeyRef:
+  #     secretKeyRef:
-        name: smtp-credentials
+  #       name: smtp-credentials
-        key: app_password
+  #       key: app_password
-  - name: EMAIL_SMTP_SERVER
+  # - name: EMAIL_SMTP_SERVER
-    value: "smtp.gmail.com"
+  #   value: "smtp.gmail.com"
-  - name: EMAIL_SMTP_PORT
+  # - name: EMAIL_SMTP_PORT
-    value: "587"
+  #   value: "587"
 
   # Application variables
   - name: POSTGRES_HOST