From 61d4b23e6933e296d4c441e0198ff9db67bce8c1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 10:02:47 -0300 Subject: [PATCH 01/10] SIENTIAPDE-1445 SIENTIAPDE-1273 Add base_scouter and pi_web_api_scouter functions to orchestrator utilities for enhanced configuration management. Update tests to validate new functionalities and ensure proper integration with existing workflows. --- orchestrator/utils/orchestrator_functions.py | 68 +++++- .../utils/test_orchestrator_functions.py | 195 ++++++++++++++++++ 2 files changed, 252 insertions(+), 11 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index b8e7eec..4f2b806 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/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]): """ @@ -110,6 +112,26 @@ def minimal_retrain(config: dict[str, Any]): 'datetime_columns': config.get('datetime_columns', []), } +def base_scouter(config: dict[str, Any]): + """ + Base scouter configuration. + """ + + 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]): """ @@ -131,9 +153,7 @@ def scouter(config: dict[str, Any]): Returns: dict[str, Any]: Scouter configuration with topic, filters, tags, and retention settings. """ - filters = {} - for f in config.get('filters', []): - filters[f['filter_name']] = {'policy': f['policy']} + tags = {} for tag in config['read_tags']: @@ -143,16 +163,42 @@ 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. + """ + tags = {} + for tag in config['read_tags']: + tags[tag['tag_name']] = { + 'webid': tag['webid'], + 'aggr_func': tag.get('aggr_func', 'lts'), + 'data_range': tag.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, + } } diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 4db42a8..efcfe86 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/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, @@ -403,3 +405,196 @@ 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': [ + { + 'tag_name': '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': [ + { + 'tag_name': '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': [ + { + 'tag_name': '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 From c98469efd20e8e1679d7fc95189accf0f98cdff2 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 10:55:29 -0300 Subject: [PATCH 02/10] SIENTIAPDE-1445 Update GITHUB_BRANCH to feature/SIENTIAPDE-1445 and enhance formatters with a new registry for workflow types, improving configuration management. Refactor process_schedules to utilize the new registry and add error handling for unsupported workflow types. Update tests to validate new functionality and ensure proper integration. --- orchestrator/activities/formatters.py | 118 +++++++++----- orchestrator/utils/orchestrator_functions.py | 153 +++++++++++++----- .../activities/test_formatters.py | 140 +++++++++++++--- .../utils/test_orchestrator_functions.py | 22 ++- values.yaml | 2 +- 5 files changed, 323 insertions(+), 112 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 9b7ce92..636d8a9 100644 --- a/orchestrator/activities/formatters.py +++ b/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,41 +178,22 @@ class Formatters(SientiaMonitoring): } for pipeline in pipelines: - if pipeline['workflow_type'] == 'scouter': - schedule_config[self.scouter_namespace][pipeline['schedule_name']] = { - **scouter(pipeline), - 'updated_at': pipeline.get( - 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) - ), - } - elif pipeline['workflow_type'] == 'predictions_batch': - schedule_config[self.laborious_namespace][pipeline['schedule_name']] = { - **predictions_batch(pipeline), - 'updated_at': pipeline.get( - 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) - ), - } - 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) - ), - } + workflow_type = pipeline['workflow_type'] + + if workflow_type not in schedule_types: + self.error(f'Workflow type {workflow_type} not supported', metadata=metadata) + continue + + schedule_type = schedule_types[workflow_type] + namespace = schedule_type['namespace'] + function = schedule_type['function'] + + schedule_config[namespace][pipeline['schedule_name']] = { + **function(pipeline), + 'updated_at': pipeline.get( + 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) + ), + } self.info('Processed schedules', metadata=metadata) self.debug(json.dumps(schedule_config, indent=4, sort_keys=True), metadata=metadata) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 4f2b806..a2c5e16 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -9,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') - - schedule_name (str): Name of the schedule - - frequency (str, optional): Frequency of execution (default: '1m') - - max_retry_policy (int, optional): Maximum retry attempts (default: 1) - - model_id (str): ID of the model - - models (dict): Model configuration containing 'name' field + - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch', 'drift') + - schedule_name (str): Unique name identifier for this schedule + - model_id (str): MongoDB ID of the associated model + - model (dict): Model configuration containing: + - name (str): Human-readable name of the model + - 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 { @@ -43,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) - - drift_metrics (list[str], optional): List of drift metrics to compute - (default: ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']) + - interval_minutes (int, optional): Time window for data comparison in minutes (default: 60) + - drift_metrics (list[str], optional): Statistical metrics to compute (default: + ['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), @@ -78,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), @@ -97,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 for retraining + - query (str): SQL query to retrieve training data. Should return features + 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), @@ -112,9 +125,28 @@ def minimal_retrain(config: dict[str, Any]): 'datetime_columns': config.get('datetime_columns', []), } + def base_scouter(config: dict[str, Any]): """ - Base scouter configuration. + 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 = {} @@ -139,22 +171,19 @@ def scouter(config: dict[str, Any]): 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 - - policy (str): Filter policy - - tag_name (str): Name of the tag - - aggr_func (str, optional): Aggregation function (default: 'lts') - - data_range (list[int], optional): Data range limits (default: [-100, 100]) - - 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 + - tag_name (str): Name of the tag to read + - aggr_func (str, optional): Aggregation function for data collection (default: 'lts') + Common values: 'lts' (last), 'avg' (average), 'min', 'max', 'sum' + - data_range (list[int], optional): Valid data range [min, max] (default: [-100, 100]) + - Additional fields from base_scouter 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}' """ - tags = {} for tag in config['read_tags']: tags[tag['tag_name']] = { @@ -168,9 +197,32 @@ def scouter(config: dict[str, Any]): 'model_tags': tags, } + 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 in config['read_tags']: @@ -198,7 +250,7 @@ def pi_web_api_scouter(config: dict[str, Any]): 'period': pi_web_api_config.get('period', '*-1d'), 'max_count': pi_web_api_config.get('max_count', 1), 'api_timeout': config_timeout, - } + }, } @@ -227,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']: @@ -252,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: - - server_id (str): ID of the OPC server - - type (str): Tag type ('prediction' or 'confidence') - - addr (str): Tag address - - data_type (str, optional): Data type (default: 'float') - - path_priority (list[str], optional): List of path priorities (default: ["STOP", "CONTINUE", "REPEAT"]) - - input_filters (list[dict], optional): List of input filter configurations - - mlflow_transform_filters (list[dict], optional): List of MLflow transform filter configurations - - mlflow_predict_filters (list[dict], optional): List of MLflow predict filter configurations - - model_retention_minutes (int, optional): Model retention time in minutes (default: 60) + - query (str): SQL query to retrieve input data for predictions + - write_tags (list[dict]): List of OPC tag configurations for write-back with: + - server_id (str): ID of the target OPC server + - type (str): Tag type - 'prediction' (model output) or 'confidence' (prediction confidence) + - addr (str): OPC tag address/path + - data_type (str, optional): OPC data type (default: 'float') + - datetime_columns (list[str], optional): Column names to parse as datetime (default: []) + - path_priority (list[str], optional): Filter policy execution order (default: ["STOP", "CONTINUE", "REPEAT"]) + - input_filters (list[dict], optional): Input data validation filters + - 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', []): diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 51b0e08..f9df04c 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/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'}) -@patch( - 'orchestrator.activities.formatters.predictions_batch', - return_value={'test_predictions_batch': 'test_predictions_batch'}, -) -@patch( - 'orchestrator.activities.formatters.minimal_retrain', - return_value={'test_minimal_retrain': 'test_minimal_retrain'}, -) -@patch( - 'orchestrator.activities.formatters.drift', - return_value={'test_drift': 'test_drift'}, -) -@patch( - 'orchestrator.activities.formatters.simple_metrics', - return_value={'test_simple_metrics': 'test_simple_metrics'}, -) -async def test_process_schedules( - mock_simple_metrics, - mock_drift, - mock_minimal_retrain, - mock_predictions_batch, - mock_scouter, - formatters, -): +async def test_process_schedules(formatters): + mock_scouter = MagicMock(return_value={'test_scouter': 'test_scouter'}) + mock_predictions_batch = MagicMock( + return_value={'test_predictions_batch': 'test_predictions_batch'} + ) + mock_minimal_retrain = MagicMock(return_value={'test_minimal_retrain': 'test_minimal_retrain'}) + mock_drift = MagicMock(return_value={'test_drift': 'test_drift'}) + mock_simple_metrics = MagicMock(return_value={'test_simple_metrics': 'test_simple_metrics'}) + + mock_schedule_types = { + 'scouter': { + 'namespace': 'scouter', + 'function': mock_scouter, + }, + 'pi_web_api_scouter': { + 'namespace': 'scouter', + 'function': mock_scouter, + }, + 'predictions_batch': { + 'namespace': 'laborious', + 'function': mock_predictions_batch, + }, + 'minimal_retrain': { + 'namespace': 'laborious', + 'function': mock_minimal_retrain, + }, + 'drift': { + 'namespace': 'laborious', + 'function': mock_drift, + }, + 'simple_metrics': { + 'namespace': 'laborious', + 'function': mock_simple_metrics, + }, + } + input_data = { 'pipelines': [ { @@ -100,7 +114,8 @@ async def test_process_schedules( ] } - result = await formatters.process_schedules(input_data) + with patch('orchestrator.activities.formatters.schedule_types', mock_schedule_types): + result = await formatters.process_schedules(input_data) assert result == { 'scouter': { @@ -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', diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index efcfe86..f225d45 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -486,7 +486,13 @@ def test_pi_web_api_scouter(): '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]}}, + '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, @@ -496,7 +502,7 @@ def test_pi_web_api_scouter(): 'period': '*-2d', 'max_count': 5, 'api_timeout': 30, - } + }, } assert result == expected @@ -536,7 +542,9 @@ def test_pi_web_api_scouter_with_timeout_greater_than_frequency(): '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]}}, + '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, @@ -546,7 +554,7 @@ def test_pi_web_api_scouter_with_timeout_greater_than_frequency(): 'period': '*-1d', 'max_count': 1, 'api_timeout': 60, - } + }, } assert result == expected @@ -585,7 +593,9 @@ def test_pi_web_api_scouter_with_no_timeout(): '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]}}, + '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, @@ -595,6 +605,6 @@ def test_pi_web_api_scouter_with_no_timeout(): 'period': '*-1d', 'max_count': 1, 'api_timeout': 30, - } + }, } assert result == expected diff --git a/values.yaml b/values.yaml index a40c8c4..ccad655 100644 --- a/values.yaml +++ b/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" From 38684071a5eb83af44c4ccfcea9d9d375dda3316 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 11:21:55 -0300 Subject: [PATCH 03/10] SIENTIAPDE-1445 Filter gather_read_tags to process only 'scouter' workflow types and add corresponding unit tests to validate this behavior. --- orchestrator/utils/orchestrator_functions.py | 3 + .../utils/test_orchestrator_functions.py | 99 +++++++++++++++++++ 2 files changed, 102 insertions(+) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index a2c5e16..f2d39b8 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -405,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: diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index f225d45..f440f3a 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -264,6 +264,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': [ { @@ -274,6 +275,7 @@ def test_gather_read_tags(): ], }, { + 'workflow_type': 'scouter', 'schedule_name': 'test_schedule2', 'read_tags': [ { @@ -316,6 +318,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 = [ { From 4a672745c500fbfe6fe08db5b06cf2412a04928e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 11:29:33 -0300 Subject: [PATCH 04/10] SIENTIAPDE-1445 Add debug print statement in gather_read_tags function to log workflow type and read tags for better traceability. --- orchestrator/utils/orchestrator_functions.py | 1 + 1 file changed, 1 insertion(+) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index f2d39b8..395462f 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -405,6 +405,7 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: # Get all read tags from pipelines for pipeline in pipelines: + print("asdddddddddddddddddddddddddd", pipeline['workflow_type'], pipeline.get('read_tags', [])) if pipeline['workflow_type'] != 'scouter': continue From 97792e36e5719149d2066f876c41a0a68d6d7059 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 11:32:05 -0300 Subject: [PATCH 05/10] SIENTIAPDE-1445 Refactor gather_read_tags function to accept a logger parameter and replace print statements with structured logging for improved traceability. --- orchestrator/utils/orchestrator_functions.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 395462f..50d76a7 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -386,7 +386,7 @@ def predictions_batch(config: dict[str, Any]): } -def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: +def gather_read_tags(pipelines: list[dict[str, Any]], logger) -> dict[str, Any]: """ Gather all read tags from input pipelines. @@ -405,7 +405,8 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: # Get all read tags from pipelines for pipeline in pipelines: - print("asdddddddddddddddddddddddddd", pipeline['workflow_type'], pipeline.get('read_tags', [])) + logger.info( + f'Processing pipeline: {pipeline["workflow_type"]}: {pipeline.get("read_tags", [])}') if pipeline['workflow_type'] != 'scouter': continue From 381b585c7ad21ff88fc3bfcf8a2ba1bd51586ef4 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 11:34:31 -0300 Subject: [PATCH 06/10] SIENTIAPDE-1445 Refactor gather_read_tags function to remove logger parameter and associated logging statements, streamlining the function for improved clarity and focus on processing pipelines. --- orchestrator/utils/orchestrator_functions.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 50d76a7..f2d39b8 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -386,7 +386,7 @@ def predictions_batch(config: dict[str, Any]): } -def gather_read_tags(pipelines: list[dict[str, Any]], logger) -> dict[str, Any]: +def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: """ Gather all read tags from input pipelines. @@ -405,8 +405,6 @@ def gather_read_tags(pipelines: list[dict[str, Any]], logger) -> dict[str, Any]: # Get all read tags from pipelines for pipeline in pipelines: - logger.info( - f'Processing pipeline: {pipeline["workflow_type"]}: {pipeline.get("read_tags", [])}') if pipeline['workflow_type'] != 'scouter': continue From 882f3b4de7cfb919a10c66ae6e4103d188b94642 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 11:51:43 -0300 Subject: [PATCH 07/10] SIENTIAPDE-1445 Refactor values.yaml to comment out email configuration settings for improved security and clarity. Update orchestrator_functions.py to enhance tag processing by iterating over items in read_tags, improving code readability. --- orchestrator/utils/orchestrator_functions.py | 10 ++++----- values.yaml | 22 ++++++++++---------- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index f2d39b8..90386a0 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -225,11 +225,11 @@ def pi_web_api_scouter(config: dict[str, Any]): API timeout is automatically adjusted to not exceed workflow frequency. """ tags = {} - for tag in config['read_tags']: - tags[tag['tag_name']] = { - 'webid': tag['webid'], - 'aggr_func': tag.get('aggr_func', 'lts'), - 'data_range': tag.get('data_range', [-100, 100]), + for tag, config in config['read_tags'].items(): + tags[tag] = { + 'webid': config['webid'], + 'aggr_func': config.get('aggr_func', 'lts'), + 'data_range': config.get('data_range', [-100, 100]), } base_config = base_scouter(config) diff --git a/values.yaml b/values.yaml index ccad655..fb697b8 100644 --- a/values.yaml +++ b/values.yaml @@ -182,17 +182,17 @@ env: - name: MONGODB_TTL_INDEX_HOURS value: "1" - - name: EMAIL_SENDER - value: "vitor.santos@aignosi.com.br" - - name: EMAIL_SENDER_PASSWORD - valueFrom: - secretKeyRef: - name: smtp-credentials - key: app_password - - name: EMAIL_SMTP_SERVER - value: "smtp.gmail.com" - - name: EMAIL_SMTP_PORT - value: "587" + # - name: EMAIL_SENDER + # value: "vitor.santos@aignosi.com.br" + # - name: EMAIL_SENDER_PASSWORD + # valueFrom: + # secretKeyRef: + # name: smtp-credentials + # key: app_password + # - name: EMAIL_SMTP_SERVER + # value: "smtp.gmail.com" + # - name: EMAIL_SMTP_PORT + # value: "587" # Application variables - name: POSTGRES_HOST From 5d2d297aa2c470c478f20eb3c71d60f9479781fd Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 11:58:14 -0300 Subject: [PATCH 08/10] SIENTIAPDE-1445 Refactor tag processing in pi_web_api_scouter function to improve clarity by renaming variables and enhancing readability of the configuration handling. --- orchestrator/utils/orchestrator_functions.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 90386a0..8b7c509 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -225,11 +225,11 @@ def pi_web_api_scouter(config: dict[str, Any]): API timeout is automatically adjusted to not exceed workflow frequency. """ tags = {} - for tag, config in config['read_tags'].items(): + for tag, tag_config in config['read_tags'].items(): tags[tag] = { - 'webid': config['webid'], - 'aggr_func': config.get('aggr_func', 'lts'), - 'data_range': config.get('data_range', [-100, 100]), + '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) From 886c6362ce6bd4f9433098d6f128db33963b4777 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 12:16:37 -0300 Subject: [PATCH 09/10] SIENTIAPDE-1445 Enhance MongoDB activity to handle missing topic configurations gracefully by using get() method and skipping empty collections, improving robustness in pipeline processing. --- orchestrator/activities/mongo_db.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 0339ab5..4ae3a96 100644 --- a/orchestrator/activities/mongo_db.py +++ b/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 From 23c6b0686841ce6e764687c9bd0c68f67cca26fb Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 22 Dec 2025 14:45:14 -0300 Subject: [PATCH 10/10] SIENTIAPDE-1445 Refactor test cases in test_orchestrator_functions.py to consolidate read_tags structure and enhance clarity in API error handling, improving test robustness and readability. --- .../utils/test_orchestrator_functions.py | 31 +++++++++++-------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index f440f3a..805a21e 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -221,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'])] @@ -553,14 +561,13 @@ def test_pi_web_api_scouter(): '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': [ - { - 'tag_name': 'test_tag_name', + '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', @@ -613,12 +620,11 @@ def test_pi_web_api_scouter_with_timeout_greater_than_frequency(): 'model_id': 'test_model_id', 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, 'filters': [], - 'read_tags': [ - { - 'tag_name': 'test_tag_name', + 'read_tags': { + 'test_tag_name': { 'webid': 'test_webid', } - ], + }, 'tag_retention_minutes': 10, 'pi_web_api_config': { 'endpoint': 'https://test-endpoint.com', @@ -665,12 +671,11 @@ def test_pi_web_api_scouter_with_no_timeout(): 'model_id': 'test_model_id', 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, 'filters': [], - 'read_tags': [ - { - 'tag_name': 'test_tag_name', + 'read_tags': { + 'test_tag_name': { 'webid': 'test_webid', } - ], + }, 'tag_retention_minutes': 10, 'pi_web_api_config': { 'endpoint': 'https://test-endpoint.com',