From 61d4b23e6933e296d4c441e0198ff9db67bce8c1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 10:02:47 -0300 Subject: [PATCH] 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