From c9b5e9872616b62d5e4236aab286095e47b28d59 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 14 Nov 2025 09:22:24 -0300 Subject: [PATCH 1/8] SIENTIAPDE-1273 Implement drift monitoring configuration and update formatters to support drift workflows. Add drift function to orchestrator utilities and enhance tests for drift functionality. --- orchestrator/activities/formatters.py | 10 +- orchestrator/utils/orchestrator_functions.py | 21 +- test.ipynb | 2040 ++++++++++++++++- .../activities/test_formatters.py | 19 +- .../utils/test_orchestrator_functions.py | 31 + 5 files changed, 2115 insertions(+), 6 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index cb45d57..73f8f86 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -19,6 +19,7 @@ with workflow.unsafe.imports_passed_through(): minimal_retrain, predictions_batch, scouter, + drift, ) topic_separator = '\n ========== \n' @@ -34,7 +35,7 @@ class Formatters(SientiaMonitoring): scheduled reports. Key features: - - Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain") + - Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain", "drift") - OPC slot distribution across active ingestors - Notification filtering for comprehensive scheduled reports - Group-based report filtering with ignore list support @@ -132,6 +133,13 @@ class Formatters(SientiaMonitoring): '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) + ), + } 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 02371a1..54c388b 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -32,6 +32,25 @@ def common_config(config: dict[str, Any]): } +def drift(config: dict[str, Any]): + """ + Build drift configuration from pipeline config. + """ + model = config['model'] + return { + **common_config(config), + 'schema': 'sientia_data', + 'source_table_name': 'laborious_data', + 'target_table_name': 'drift_metrics', + 'interval': config.get('interval_minutes', 60), + 'drift_metrics': config.get('drift_metrics', [ + "kolmogorov_smirnov", + "jensen_shannon", + "wasserstein" + ]) + } + + def minimal_retrain(config: dict[str, Any]): """ Build minimal retrain configuration from pipeline config. @@ -47,8 +66,6 @@ def minimal_retrain(config: dict[str, Any]): """ return { **common_config(config), - 'workflow_type': 'minimal_retrain', - 'schedule_name': config['schedule_name'], 'query': config['query'], 'schema': 'sientia_data', 'table_name': 'log_retrain', diff --git a/test.ipynb b/test.ipynb index 1843727..a101c3c 100644 --- a/test.ipynb +++ b/test.ipynb @@ -617,13 +617,2049 @@ " print(f\"Chave: {key}, Valor: {value}\")" ] }, + { + "cell_type": "markdown", + "id": "e8025cbb", + "metadata": {}, + "source": [ + "## Model Monitoring" + ] + }, { "cell_type": "code", - "execution_count": null, + "execution_count": 1, "id": "87a9dc89", "metadata": {}, "outputs": [], - "source": [] + "source": [ + "from sientia.ModelAnalysis import ModelAnalysis\n", + "\n", + "config = {\n", + " \"target\": \"Square\",\n", + " \"prediction\": \"prediction\",\n", + " \"timestamp\": \"timestamp\",\n", + " \"features\": [\"Counter\", \"Rollout\"]\n", + "}\n", + "\n", + "model = ModelAnalysis(config=config)\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "id": "2a74ad2f", + "metadata": {}, + "outputs": [], + "source": [ + "from sientia_do.temporal.activities.postgres import Postgres\n", + "from unittest.mock import AsyncMock\n", + "from pandas import DataFrame\n", + "\n", + "postgres = Postgres(\n", + " host=\"localhost\",\n", + " port=5432,\n", + " dbname=\"sientia\",\n", + " user=\"sientia\",\n", + " password=\"sientia\",\n", + " min_connections=1,\n", + " max_connections=10,\n", + " logger=AsyncMock(),\n", + " notification_handler=AsyncMock(),\n", + " metrics_controller=AsyncMock()\n", + ")\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "938cecbd", + "metadata": {}, + "outputs": [ + { + "name": "stderr", + "output_type": "stream", + "text": [ + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/sientia_do/observability/sientia_monitoring.py:83: RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited\n", + " self.logger.custom_info(message, metadata)\n", + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/sientia_do/observability/sientia_monitoring.py:59: RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited\n", + " self.metrics_controller.start()\n", + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/sientia_do/observability/sientia_monitoring.py:83: RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited\n", + " self.logger.custom_info(message, metadata)\n", + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/sientia_do/observability/sientia_monitoring.py:95: RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited\n", + " self.logger.custom_debug(message, metadata)\n" + ] + }, + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
variablevaluepredictiontimestamp
0Square42.2950-3.0184652025-11-12 12:44:50+00:00
1Rollout1.8610-3.0184652025-11-12 12:44:50+00:00
2Counter-45.0980-3.0184652025-11-12 12:44:50+00:00
3Rollout0.39702.3414232025-11-12 12:44:20+00:00
4Counter-40.02602.3414232025-11-12 12:44:20+00:00
5Square43.31902.3414232025-11-12 12:44:20+00:00
6Rollout-0.26454.6965912025-11-12 12:42:15+00:00
7Square42.18504.6965912025-11-12 12:42:15+00:00
8Counter-37.83854.6965912025-11-12 12:42:15+00:00
9Square2.8880-70.8451572025-11-11 23:00:48+00:00
10Rollout18.7945-70.8451572025-11-11 23:00:48+00:00
11Counter-12.6565-70.8451572025-11-11 23:00:48+00:00
12Rollout17.7415-71.8945902025-11-11 23:00:18+00:00
13Counter-11.1570-71.8945902025-11-11 23:00:18+00:00
14Square2.1610-71.8945902025-11-11 23:00:18+00:00
15Square-2.0490-67.0339532025-11-11 22:59:48+00:00
16Rollout16.5365-67.0339532025-11-11 22:59:48+00:00
17Counter-15.2205-67.0339532025-11-11 22:59:48+00:00
18Counter-11.7930-70.5144622025-11-11 22:59:18+00:00
19Square-4.3350-70.5144622025-11-11 22:59:18+00:00
20Rollout16.3065-70.5144622025-11-11 22:59:18+00:00
21Square0.6840-72.5232792025-11-11 22:58:48+00:00
22Counter-9.2570-72.5232792025-11-11 22:58:48+00:00
23Rollout14.9965-72.5232792025-11-11 22:58:48+00:00
24Square1.1700-72.8316842025-11-11 22:58:18+00:00
25Counter-7.0635-72.8316842025-11-11 22:58:18+00:00
26Rollout10.9875-72.8316842025-11-11 22:58:18+00:00
27Counter-2.0110-77.8659442025-11-11 22:57:48+00:00
28Rollout10.4545-77.8659442025-11-11 22:57:48+00:00
29Square-0.2070-77.8659442025-11-11 22:57:48+00:00
\n", + "
" + ], + "text/plain": [ + " variable value prediction timestamp\n", + "0 Square 42.2950 -3.018465 2025-11-12 12:44:50+00:00\n", + "1 Rollout 1.8610 -3.018465 2025-11-12 12:44:50+00:00\n", + "2 Counter -45.0980 -3.018465 2025-11-12 12:44:50+00:00\n", + "3 Rollout 0.3970 2.341423 2025-11-12 12:44:20+00:00\n", + "4 Counter -40.0260 2.341423 2025-11-12 12:44:20+00:00\n", + "5 Square 43.3190 2.341423 2025-11-12 12:44:20+00:00\n", + "6 Rollout -0.2645 4.696591 2025-11-12 12:42:15+00:00\n", + "7 Square 42.1850 4.696591 2025-11-12 12:42:15+00:00\n", + "8 Counter -37.8385 4.696591 2025-11-12 12:42:15+00:00\n", + "9 Square 2.8880 -70.845157 2025-11-11 23:00:48+00:00\n", + "10 Rollout 18.7945 -70.845157 2025-11-11 23:00:48+00:00\n", + "11 Counter -12.6565 -70.845157 2025-11-11 23:00:48+00:00\n", + "12 Rollout 17.7415 -71.894590 2025-11-11 23:00:18+00:00\n", + "13 Counter -11.1570 -71.894590 2025-11-11 23:00:18+00:00\n", + "14 Square 2.1610 -71.894590 2025-11-11 23:00:18+00:00\n", + "15 Square -2.0490 -67.033953 2025-11-11 22:59:48+00:00\n", + "16 Rollout 16.5365 -67.033953 2025-11-11 22:59:48+00:00\n", + "17 Counter -15.2205 -67.033953 2025-11-11 22:59:48+00:00\n", + "18 Counter -11.7930 -70.514462 2025-11-11 22:59:18+00:00\n", + "19 Square -4.3350 -70.514462 2025-11-11 22:59:18+00:00\n", + "20 Rollout 16.3065 -70.514462 2025-11-11 22:59:18+00:00\n", + "21 Square 0.6840 -72.523279 2025-11-11 22:58:48+00:00\n", + "22 Counter -9.2570 -72.523279 2025-11-11 22:58:48+00:00\n", + "23 Rollout 14.9965 -72.523279 2025-11-11 22:58:48+00:00\n", + "24 Square 1.1700 -72.831684 2025-11-11 22:58:18+00:00\n", + "25 Counter -7.0635 -72.831684 2025-11-11 22:58:18+00:00\n", + "26 Rollout 10.9875 -72.831684 2025-11-11 22:58:18+00:00\n", + "27 Counter -2.0110 -77.865944 2025-11-11 22:57:48+00:00\n", + "28 Rollout 10.4545 -77.865944 2025-11-11 22:57:48+00:00\n", + "29 Square -0.2070 -77.865944 2025-11-11 22:57:48+00:00" + ] + }, + "metadata": {}, + "output_type": "display_data" + }, + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
variablevaluepredictiontimestamp
0Square42.2950-3.0184652025-11-12 12:44:50
1Rollout1.8610-3.0184652025-11-12 12:44:50
2Counter-45.0980-3.0184652025-11-12 12:44:50
3Rollout0.39702.3414232025-11-12 12:44:20
4Counter-40.02602.3414232025-11-12 12:44:20
5Square43.31902.3414232025-11-12 12:44:20
6Rollout-0.26454.6965912025-11-12 12:42:15
7Square42.18504.6965912025-11-12 12:42:15
8Counter-37.83854.6965912025-11-12 12:42:15
9Square2.8880-70.8451572025-11-11 23:00:48
10Rollout18.7945-70.8451572025-11-11 23:00:48
11Counter-12.6565-70.8451572025-11-11 23:00:48
12Rollout17.7415-71.8945902025-11-11 23:00:18
13Counter-11.1570-71.8945902025-11-11 23:00:18
14Square2.1610-71.8945902025-11-11 23:00:18
15Square-2.0490-67.0339532025-11-11 22:59:48
16Rollout16.5365-67.0339532025-11-11 22:59:48
17Counter-15.2205-67.0339532025-11-11 22:59:48
18Counter-11.7930-70.5144622025-11-11 22:59:18
19Square-4.3350-70.5144622025-11-11 22:59:18
20Rollout16.3065-70.5144622025-11-11 22:59:18
21Square0.6840-72.5232792025-11-11 22:58:48
22Counter-9.2570-72.5232792025-11-11 22:58:48
23Rollout14.9965-72.5232792025-11-11 22:58:48
24Square1.1700-72.8316842025-11-11 22:58:18
25Counter-7.0635-72.8316842025-11-11 22:58:18
26Rollout10.9875-72.8316842025-11-11 22:58:18
27Counter-2.0110-77.8659442025-11-11 22:57:48
28Rollout10.4545-77.8659442025-11-11 22:57:48
29Square-0.2070-77.8659442025-11-11 22:57:48
\n", + "
" + ], + "text/plain": [ + " variable value prediction timestamp\n", + "0 Square 42.2950 -3.018465 2025-11-12 12:44:50\n", + "1 Rollout 1.8610 -3.018465 2025-11-12 12:44:50\n", + "2 Counter -45.0980 -3.018465 2025-11-12 12:44:50\n", + "3 Rollout 0.3970 2.341423 2025-11-12 12:44:20\n", + "4 Counter -40.0260 2.341423 2025-11-12 12:44:20\n", + "5 Square 43.3190 2.341423 2025-11-12 12:44:20\n", + "6 Rollout -0.2645 4.696591 2025-11-12 12:42:15\n", + "7 Square 42.1850 4.696591 2025-11-12 12:42:15\n", + "8 Counter -37.8385 4.696591 2025-11-12 12:42:15\n", + "9 Square 2.8880 -70.845157 2025-11-11 23:00:48\n", + "10 Rollout 18.7945 -70.845157 2025-11-11 23:00:48\n", + "11 Counter -12.6565 -70.845157 2025-11-11 23:00:48\n", + "12 Rollout 17.7415 -71.894590 2025-11-11 23:00:18\n", + "13 Counter -11.1570 -71.894590 2025-11-11 23:00:18\n", + "14 Square 2.1610 -71.894590 2025-11-11 23:00:18\n", + "15 Square -2.0490 -67.033953 2025-11-11 22:59:48\n", + "16 Rollout 16.5365 -67.033953 2025-11-11 22:59:48\n", + "17 Counter -15.2205 -67.033953 2025-11-11 22:59:48\n", + "18 Counter -11.7930 -70.514462 2025-11-11 22:59:18\n", + "19 Square -4.3350 -70.514462 2025-11-11 22:59:18\n", + "20 Rollout 16.3065 -70.514462 2025-11-11 22:59:18\n", + "21 Square 0.6840 -72.523279 2025-11-11 22:58:48\n", + "22 Counter -9.2570 -72.523279 2025-11-11 22:58:48\n", + "23 Rollout 14.9965 -72.523279 2025-11-11 22:58:48\n", + "24 Square 1.1700 -72.831684 2025-11-11 22:58:18\n", + "25 Counter -7.0635 -72.831684 2025-11-11 22:58:18\n", + "26 Rollout 10.9875 -72.831684 2025-11-11 22:58:18\n", + "27 Counter -2.0110 -77.865944 2025-11-11 22:57:48\n", + "28 Rollout 10.4545 -77.865944 2025-11-11 22:57:48\n", + "29 Square -0.2070 -77.865944 2025-11-11 22:57:48" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "from pandas import DataFrame, to_datetime\n", + "\n", + "query_data = \"\"\"\n", + "select ld.variable, ld.value, p.prediction, p.\"timestamp\"\n", + "from sientia_data.laborious_data ld\n", + " right join sientia_data.predictions p \n", + " on ld.timestamp = p.timestamp\n", + " and ld.model_id = p.model_id\n", + " where\n", + " ld.model_id = '1'\n", + " order by\n", + " ld.created_at desc limit 30;\n", + "\"\"\"\n", + "\n", + "input_data = {\n", + " \"query\": query_data,\n", + " \"metadata\": {}\n", + "}\n", + "\n", + "\n", + "raw_data = DataFrame(await postgres.load_custom_query(\n", + " input_data))\n", + "\n", + "raw_data['timestamp'] = to_datetime(raw_data['timestamp'])\n", + "raw_data['timestamp'] = raw_data['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S')\n", + "\n", + "display(raw_data)" + ] + }, + { + "cell_type": "code", + "execution_count": 7, + "id": "297f648f", + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
timestamppredictionCounterRolloutSquare
02025-11-11 22:57:48-77.865944-2.011010.4545-0.207
12025-11-11 22:58:18-72.831684-7.063510.98751.170
22025-11-11 22:58:48-72.523279-9.257014.99650.684
32025-11-11 22:59:18-70.514462-11.793016.3065-4.335
42025-11-11 22:59:48-67.033953-15.220516.5365-2.049
52025-11-11 23:00:18-71.894590-11.157017.74152.161
62025-11-11 23:00:48-70.845157-12.656518.79452.888
72025-11-12 12:42:154.696591-37.8385-0.264542.185
82025-11-12 12:44:202.341423-40.02600.397043.319
92025-11-12 12:44:50-3.018465-45.09801.861042.295
\n", + "
" + ], + "text/plain": [ + " timestamp prediction Counter Rollout Square\n", + "0 2025-11-11 22:57:48 -77.865944 -2.0110 10.4545 -0.207\n", + "1 2025-11-11 22:58:18 -72.831684 -7.0635 10.9875 1.170\n", + "2 2025-11-11 22:58:48 -72.523279 -9.2570 14.9965 0.684\n", + "3 2025-11-11 22:59:18 -70.514462 -11.7930 16.3065 -4.335\n", + "4 2025-11-11 22:59:48 -67.033953 -15.2205 16.5365 -2.049\n", + "5 2025-11-11 23:00:18 -71.894590 -11.1570 17.7415 2.161\n", + "6 2025-11-11 23:00:48 -70.845157 -12.6565 18.7945 2.888\n", + "7 2025-11-12 12:42:15 4.696591 -37.8385 -0.2645 42.185\n", + "8 2025-11-12 12:44:20 2.341423 -40.0260 0.3970 43.319\n", + "9 2025-11-12 12:44:50 -3.018465 -45.0980 1.8610 42.295" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "from pandas import merge, pivot\n", + "\n", + "raw_data.drop_duplicates(subset=[\"timestamp\", \"prediction\", \"variable\"], inplace=True, keep=\"first\")\n", + "data = raw_data.pivot(index=[\"timestamp\", \"prediction\"], columns=\"variable\", values=\"value\")\n", + "\n", + "data.columns.name = None\n", + "data.reset_index(inplace=True)\n", + "display(data)\n", + "\n", + "\n", + "drift_methods = ['kolmogorov_smirnov']#, 'jensen_shannon', 'wasserstein']\n", + "chunk_period = 's'\n", + "target_col = \"Square\"\n", + "timestamp_col = \"timestamp\"\n", + "\n", + "data.to_csv(\"data.csv\", index=False)\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": 10, + "id": "9ab4774a", + "metadata": {}, + "outputs": [], + "source": [ + "from mlflow.tracking import MlflowClient\n", + "from os import environ\n", + "\n", + "environ['MLFLOW_TRACKING_USERNAME'] = 'aignosi'\n", + "environ['MLFLOW_TRACKING_PASSWORD'] = '1L0FP50j3ncp123'\n", + "environ['MLFLOW_TRACKING_URI'] = 'http://localhost:5080'\n", + "mlflow_client = MlflowClient()" + ] + }, + { + "cell_type": "code", + "execution_count": 19, + "id": "e8427625", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "True" + ] + }, + "execution_count": 19, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "run_id = '9e9fa748822c44ab92dd769d6a45c4f4'\n", + "output_dir = './tmp/model'\n", + "\n", + "artifacts = mlflow_client.list_artifacts(run_id)\n", + "paths = [artifact.path for artifact in artifacts]\n", + "\n", + "any(artifact.path == 'retrain_data.csv' for artifact in artifacts)" + ] + }, + { + "cell_type": "code", + "execution_count": 12, + "id": "cf4ddf41", + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
timestampCounterRolloutSquare
02025-11-10 18:58:3913.5970-52.450043.381
12025-11-10 18:58:4915.0660-50.024045.813
22025-11-10 18:58:5414.2780-49.481047.677
32025-11-10 18:59:0414.2080-51.801548.995
42025-11-10 18:59:0916.3530-52.685047.745
52025-11-10 18:59:1915.5415-53.562548.327
62025-11-10 18:59:2416.8580-53.696047.160
72025-11-10 18:59:3414.9745-54.751048.480
82025-11-10 18:59:3912.3020-53.433048.631
92025-11-10 18:59:4912.5640-52.980550.130
\n", + "
" + ], + "text/plain": [ + " timestamp Counter Rollout Square\n", + "0 2025-11-10 18:58:39 13.5970 -52.4500 43.381\n", + "1 2025-11-10 18:58:49 15.0660 -50.0240 45.813\n", + "2 2025-11-10 18:58:54 14.2780 -49.4810 47.677\n", + "3 2025-11-10 18:59:04 14.2080 -51.8015 48.995\n", + "4 2025-11-10 18:59:09 16.3530 -52.6850 47.745\n", + "5 2025-11-10 18:59:19 15.5415 -53.5625 48.327\n", + "6 2025-11-10 18:59:24 16.8580 -53.6960 47.160\n", + "7 2025-11-10 18:59:34 14.9745 -54.7510 48.480\n", + "8 2025-11-10 18:59:39 12.3020 -53.4330 48.631\n", + "9 2025-11-10 18:59:49 12.5640 -52.9805 50.130" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "import mlflow\n", + "from io import StringIO\n", + "from pandas import read_csv\n", + "\n", + "if 'retrain_data.csv' in paths:\n", + " artifact_path = 'retrain_data.csv'\n", + "elif 'train_data.csv' in paths:\n", + " artifact_path = 'train_data.csv'\n", + "else:\n", + " raise RuntimeError(\n", + " f\"Could not find 'train_data.csv' or 'retrain_data.csv' for run_id '{run_id}'.\\n\"\n", + " )\n", + "\n", + "# Carregar diretamente na memória como string\n", + "artifact_content = mlflow.artifacts.load_text(\n", + " f\"runs:/{run_id}/{artifact_path}\"\n", + ")\n", + "\n", + "reference_df = read_csv(StringIO(artifact_content))\n", + "\n", + "reference_df.drop(columns=['timestamp.1'], inplace=True)\n", + "\n", + "reference_df['timestamp'] = to_datetime(reference_df['timestamp'])\n", + "reference_df['timestamp'] = reference_df['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S')\n", + "\n", + "display(reference_df)\n", + "\n", + "reference_df.to_csv(\"reference_df.csv\", index=False)" + ] + }, + { + "cell_type": "code", + "execution_count": 20, + "id": "d375752b", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "\n" + ] + }, + { + "name": "stderr", + "output_type": "stream", + "text": [ + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/numpy/core/_methods.py:206: RuntimeWarning: Degrees of freedom <= 0 for slice\n", + " ret = _var(a, axis=axis, dtype=dtype, out=out, ddof=ddof,\n", + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/numpy/core/_methods.py:198: RuntimeWarning: invalid value encountered in scalar divide\n", + " ret = ret.dtype.type(ret / rcount)\n", + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/numpy/core/_methods.py:206: RuntimeWarning: Degrees of freedom <= 0 for slice\n", + " ret = _var(a, axis=axis, dtype=dtype, out=out, ddof=ddof,\n", + "/home/grezewave/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/numpy/core/_methods.py:198: RuntimeWarning: invalid value encountered in scalar divide\n", + " ret = ret.dtype.type(ret / rcount)\n" + ] + }, + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
timestampfeaturemetricstatisticp_valuealertchunk_indexchunk_start_datechunk_end_date
02025-11-10 18:58:00+0000Counterkolmogorov_smirnov3.000000e-01NaNFalse02025-11-10 18:582025-11-10 18:58:59.999999999
12025-11-10 18:59:00+0000Counterkolmogorov_smirnov1.285714e-01NaNFalse12025-11-10 18:592025-11-10 18:59:59.999999999
22025-11-11 22:57:00+0000Counterkolmogorov_smirnov1.000000e+00NaNTrue02025-11-11 22:572025-11-11 22:57:59.999999999
32025-11-11 22:58:00+0000Counterkolmogorov_smirnov1.000000e+00NaNTrue12025-11-11 22:582025-11-11 22:58:59.999999999
42025-11-11 22:59:00+0000Counterkolmogorov_smirnov1.000000e+00NaNTrue22025-11-11 22:592025-11-11 22:59:59.999999999
52025-11-11 23:00:00+0000Counterkolmogorov_smirnov1.000000e+00NaNTrue32025-11-11 23:002025-11-11 23:00:59.999999999
62025-11-12 12:42:00+0000Counterkolmogorov_smirnov1.000000e+00NaNTrue42025-11-12 12:422025-11-12 12:42:59.999999999
72025-11-12 12:44:00+0000Counterkolmogorov_smirnov1.000000e+00NaNTrue52025-11-12 12:442025-11-12 12:44:59.999999999
82025-11-10 18:58:00+0000Rolloutkolmogorov_smirnov6.000000e-01NaNFalse02025-11-10 18:582025-11-10 18:58:59.999999999
92025-11-10 18:59:00+0000Rolloutkolmogorov_smirnov2.571429e-01NaNFalse12025-11-10 18:592025-11-10 18:59:59.999999999
102025-11-11 22:57:00+0000Rolloutkolmogorov_smirnov1.000000e+00NaNTrue02025-11-11 22:572025-11-11 22:57:59.999999999
112025-11-11 22:58:00+0000Rolloutkolmogorov_smirnov1.000000e+00NaNTrue12025-11-11 22:582025-11-11 22:58:59.999999999
122025-11-11 22:59:00+0000Rolloutkolmogorov_smirnov1.000000e+00NaNTrue22025-11-11 22:592025-11-11 22:59:59.999999999
132025-11-11 23:00:00+0000Rolloutkolmogorov_smirnov1.000000e+00NaNTrue32025-11-11 23:002025-11-11 23:00:59.999999999
142025-11-12 12:42:00+0000Rolloutkolmogorov_smirnov1.000000e+00NaNTrue42025-11-12 12:422025-11-12 12:42:59.999999999
152025-11-12 12:44:00+0000Rolloutkolmogorov_smirnov1.000000e+00NaNTrue52025-11-12 12:442025-11-12 12:44:59.999999999
162025-11-10 18:58:00+0000multivariatemultivariate2.015043e-163.748123e-17False02025-11-10 18:582025-11-10 18:58:59.999999999
172025-11-10 18:59:00+0000multivariatemultivariate2.493220e-163.998584e-17False12025-11-10 18:592025-11-10 18:59:59.999999999
182025-11-11 22:57:00+0000multivariatemultivariate1.432145e-14NaNTrue02025-11-11 22:572025-11-11 22:57:59.999999999
192025-11-11 22:58:00+0000multivariatemultivariate1.234840e-142.299811e-15True12025-11-11 22:582025-11-11 22:58:59.999999999
202025-11-11 22:59:00+0000multivariatemultivariate7.944109e-150.000000e+00True22025-11-11 22:592025-11-11 22:59:59.999999999
212025-11-11 23:00:00+0000multivariatemultivariate7.524768e-154.193410e-16True32025-11-11 23:002025-11-11 23:00:59.999999999
222025-11-12 12:42:00+0000multivariatemultivariate1.004859e-14NaNTrue42025-11-12 12:422025-11-12 12:42:59.999999999
232025-11-12 12:44:00+0000multivariatemultivariate1.004859e-140.000000e+00True52025-11-12 12:442025-11-12 12:44:59.999999999
\n", + "
" + ], + "text/plain": [ + " timestamp feature metric statistic \\\n", + "0 2025-11-10 18:58:00+0000 Counter kolmogorov_smirnov 3.000000e-01 \n", + "1 2025-11-10 18:59:00+0000 Counter kolmogorov_smirnov 1.285714e-01 \n", + "2 2025-11-11 22:57:00+0000 Counter kolmogorov_smirnov 1.000000e+00 \n", + "3 2025-11-11 22:58:00+0000 Counter kolmogorov_smirnov 1.000000e+00 \n", + "4 2025-11-11 22:59:00+0000 Counter kolmogorov_smirnov 1.000000e+00 \n", + "5 2025-11-11 23:00:00+0000 Counter kolmogorov_smirnov 1.000000e+00 \n", + "6 2025-11-12 12:42:00+0000 Counter kolmogorov_smirnov 1.000000e+00 \n", + "7 2025-11-12 12:44:00+0000 Counter kolmogorov_smirnov 1.000000e+00 \n", + "8 2025-11-10 18:58:00+0000 Rollout kolmogorov_smirnov 6.000000e-01 \n", + "9 2025-11-10 18:59:00+0000 Rollout kolmogorov_smirnov 2.571429e-01 \n", + "10 2025-11-11 22:57:00+0000 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "11 2025-11-11 22:58:00+0000 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "12 2025-11-11 22:59:00+0000 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "13 2025-11-11 23:00:00+0000 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "14 2025-11-12 12:42:00+0000 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "15 2025-11-12 12:44:00+0000 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "16 2025-11-10 18:58:00+0000 multivariate multivariate 2.015043e-16 \n", + "17 2025-11-10 18:59:00+0000 multivariate multivariate 2.493220e-16 \n", + "18 2025-11-11 22:57:00+0000 multivariate multivariate 1.432145e-14 \n", + "19 2025-11-11 22:58:00+0000 multivariate multivariate 1.234840e-14 \n", + "20 2025-11-11 22:59:00+0000 multivariate multivariate 7.944109e-15 \n", + "21 2025-11-11 23:00:00+0000 multivariate multivariate 7.524768e-15 \n", + "22 2025-11-12 12:42:00+0000 multivariate multivariate 1.004859e-14 \n", + "23 2025-11-12 12:44:00+0000 multivariate multivariate 1.004859e-14 \n", + "\n", + " p_value alert chunk_index chunk_start_date \\\n", + "0 NaN False 0 2025-11-10 18:58 \n", + "1 NaN False 1 2025-11-10 18:59 \n", + "2 NaN True 0 2025-11-11 22:57 \n", + "3 NaN True 1 2025-11-11 22:58 \n", + "4 NaN True 2 2025-11-11 22:59 \n", + "5 NaN True 3 2025-11-11 23:00 \n", + "6 NaN True 4 2025-11-12 12:42 \n", + "7 NaN True 5 2025-11-12 12:44 \n", + "8 NaN False 0 2025-11-10 18:58 \n", + "9 NaN False 1 2025-11-10 18:59 \n", + "10 NaN True 0 2025-11-11 22:57 \n", + "11 NaN True 1 2025-11-11 22:58 \n", + "12 NaN True 2 2025-11-11 22:59 \n", + "13 NaN True 3 2025-11-11 23:00 \n", + "14 NaN True 4 2025-11-12 12:42 \n", + "15 NaN True 5 2025-11-12 12:44 \n", + "16 3.748123e-17 False 0 2025-11-10 18:58 \n", + "17 3.998584e-17 False 1 2025-11-10 18:59 \n", + "18 NaN True 0 2025-11-11 22:57 \n", + "19 2.299811e-15 True 1 2025-11-11 22:58 \n", + "20 0.000000e+00 True 2 2025-11-11 22:59 \n", + "21 4.193410e-16 True 3 2025-11-11 23:00 \n", + "22 NaN True 4 2025-11-12 12:42 \n", + "23 0.000000e+00 True 5 2025-11-12 12:44 \n", + "\n", + " chunk_end_date \n", + "0 2025-11-10 18:58:59.999999999 \n", + "1 2025-11-10 18:59:59.999999999 \n", + "2 2025-11-11 22:57:59.999999999 \n", + "3 2025-11-11 22:58:59.999999999 \n", + "4 2025-11-11 22:59:59.999999999 \n", + "5 2025-11-11 23:00:59.999999999 \n", + "6 2025-11-12 12:42:59.999999999 \n", + "7 2025-11-12 12:44:59.999999999 \n", + "8 2025-11-10 18:58:59.999999999 \n", + "9 2025-11-10 18:59:59.999999999 \n", + "10 2025-11-11 22:57:59.999999999 \n", + "11 2025-11-11 22:58:59.999999999 \n", + "12 2025-11-11 22:59:59.999999999 \n", + "13 2025-11-11 23:00:59.999999999 \n", + "14 2025-11-12 12:42:59.999999999 \n", + "15 2025-11-12 12:44:59.999999999 \n", + "16 2025-11-10 18:58:59.999999999 \n", + "17 2025-11-10 18:59:59.999999999 \n", + "18 2025-11-11 22:57:59.999999999 \n", + "19 2025-11-11 22:58:59.999999999 \n", + "20 2025-11-11 22:59:59.999999999 \n", + "21 2025-11-11 23:00:59.999999999 \n", + "22 2025-11-12 12:42:59.999999999 \n", + "23 2025-11-12 12:44:59.999999999 " + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "\n", + "\n", + "from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ\n", + "\n", + "\n", + "univariate_drift = model.detect_univariate_drift(\n", + " reference_df=reference_df,\n", + " analysis_df=data,\n", + " features=config['features'],\n", + " timestamp_col=timestamp_col,\n", + " methods=drift_methods,\n", + " chunk_period=\"min\"\n", + ")\n", + "multivariate_drift = model.detect_multivariate_drift(\n", + " reference_df=reference_df,\n", + " analysis_df=data,\n", + " features=config['features'],\n", + " timestamp_col=timestamp_col,\n", + " chunk_period=\"min\"\n", + ")\n", + "\n", + "drift = model.get_drift_metrics_dataframe(\n", + " univariate_drift=univariate_drift,\n", + " multivariate_drift=multivariate_drift\n", + ")\n", + "drift.drop_duplicates(subset=[\"timestamp\", \"feature\", \"metric\"], inplace=True, keep=\"first\")\n", + "drift.reset_index(drop=True, inplace=True)\n", + "\n", + "drift['timestamp'] = to_datetime(drift['timestamp'])\n", + "print(type(drift['timestamp'].iloc[0]))\n", + "\n", + "# Add timezone UTC to timestamp\n", + "drift['timestamp'] = drift['timestamp'].dt.tz_localize('UTC')\n", + "drift['timestamp'] = drift['timestamp'].dt.strftime(DATETIME_FORMAT_WITH_TZ)\n", + "\n", + "display(drift)\n", + "drift.to_csv(\"drift.csv\", index=False)" + ] + }, + { + "cell_type": "code", + "execution_count": 32, + "id": "d30f6475", + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
timestampfeaturemetricstatisticp_valuealertchunk_indexchunk_start_datechunk_end_date
102025-11-11 11:55:18Counterkolmogorov_smirnov1.000000e+00NoneFalse02025-11-11 11:55:182025-11-11 11:55:18.999999999
112025-11-11 11:55:48Counterkolmogorov_smirnov1.000000e+00NoneFalse12025-11-11 11:55:482025-11-11 11:55:48.999999999
122025-11-11 11:56:18Counterkolmogorov_smirnov1.000000e+00NoneFalse22025-11-11 11:56:182025-11-11 11:56:18.999999999
132025-11-11 11:56:48Counterkolmogorov_smirnov1.000000e+00NoneFalse32025-11-11 11:56:482025-11-11 11:56:48.999999999
142025-11-11 11:57:18Counterkolmogorov_smirnov1.000000e+00NoneFalse42025-11-11 11:57:182025-11-11 11:57:18.999999999
152025-11-11 11:57:48Counterkolmogorov_smirnov1.000000e+00NoneFalse52025-11-11 11:57:482025-11-11 11:57:48.999999999
162025-11-11 11:58:18Counterkolmogorov_smirnov1.000000e+00NoneFalse62025-11-11 11:58:182025-11-11 11:58:18.999999999
172025-11-11 11:58:48Counterkolmogorov_smirnov1.000000e+00NoneFalse72025-11-11 11:58:482025-11-11 11:58:48.999999999
182025-11-11 11:59:18Counterkolmogorov_smirnov1.000000e+00NoneFalse82025-11-11 11:59:182025-11-11 11:59:18.999999999
192025-11-11 11:59:48Counterkolmogorov_smirnov1.000000e+00NoneFalse92025-11-11 11:59:482025-11-11 11:59:48.999999999
302025-11-11 11:55:18Rolloutkolmogorov_smirnov1.000000e+00NoneFalse02025-11-11 11:55:182025-11-11 11:55:18.999999999
312025-11-11 11:55:48Rolloutkolmogorov_smirnov1.000000e+00NoneFalse12025-11-11 11:55:482025-11-11 11:55:48.999999999
322025-11-11 11:56:18Rolloutkolmogorov_smirnov1.000000e+00NoneFalse22025-11-11 11:56:182025-11-11 11:56:18.999999999
332025-11-11 11:56:48Rolloutkolmogorov_smirnov1.000000e+00NoneFalse32025-11-11 11:56:482025-11-11 11:56:48.999999999
342025-11-11 11:57:18Rolloutkolmogorov_smirnov1.000000e+00NoneFalse42025-11-11 11:57:182025-11-11 11:57:18.999999999
352025-11-11 11:57:48Rolloutkolmogorov_smirnov1.000000e+00NoneFalse52025-11-11 11:57:482025-11-11 11:57:48.999999999
362025-11-11 11:58:18Rolloutkolmogorov_smirnov1.000000e+00NoneFalse62025-11-11 11:58:182025-11-11 11:58:18.999999999
372025-11-11 11:58:48Rolloutkolmogorov_smirnov1.000000e+00NoneFalse72025-11-11 11:58:482025-11-11 11:58:48.999999999
382025-11-11 11:59:18Rolloutkolmogorov_smirnov1.000000e+00NoneFalse82025-11-11 11:59:182025-11-11 11:59:18.999999999
392025-11-11 11:59:48Rolloutkolmogorov_smirnov1.000000e+00NoneFalse92025-11-11 11:59:482025-11-11 11:59:48.999999999
502025-11-11 11:55:18multivariatemultivariate1.517720e-14NaNTrue02025-11-11 11:55:182025-11-11 11:55:18.999999999
512025-11-11 11:55:48multivariatemultivariate1.588822e-14NaNTrue12025-11-11 11:55:482025-11-11 11:55:48.999999999
522025-11-11 11:56:18multivariatemultivariate1.432145e-14NaNTrue22025-11-11 11:56:182025-11-11 11:56:18.999999999
532025-11-11 11:56:48multivariatemultivariate1.421085e-14NaNTrue32025-11-11 11:56:482025-11-11 11:56:48.999999999
542025-11-11 11:57:18multivariatemultivariate1.421085e-14NaNTrue42025-11-11 11:57:182025-11-11 11:57:18.999999999
552025-11-11 11:57:48multivariatemultivariate1.776357e-15NaNTrue52025-11-11 11:57:482025-11-11 11:57:48.999999999
562025-11-11 11:58:18multivariatemultivariate7.324107e-15NaNTrue62025-11-11 11:58:182025-11-11 11:58:18.999999999
572025-11-11 11:58:48multivariatemultivariate1.464821e-14NaNTrue72025-11-11 11:58:482025-11-11 11:58:48.999999999
582025-11-11 11:59:18multivariatemultivariate1.551137e-14NaNTrue82025-11-11 11:59:182025-11-11 11:59:18.999999999
592025-11-11 11:59:48multivariatemultivariate1.427317e-14NaNTrue92025-11-11 11:59:482025-11-11 11:59:48.999999999
\n", + "
" + ], + "text/plain": [ + " timestamp feature metric statistic \\\n", + "10 2025-11-11 11:55:18 Counter kolmogorov_smirnov 1.000000e+00 \n", + "11 2025-11-11 11:55:48 Counter kolmogorov_smirnov 1.000000e+00 \n", + "12 2025-11-11 11:56:18 Counter kolmogorov_smirnov 1.000000e+00 \n", + "13 2025-11-11 11:56:48 Counter kolmogorov_smirnov 1.000000e+00 \n", + "14 2025-11-11 11:57:18 Counter kolmogorov_smirnov 1.000000e+00 \n", + "15 2025-11-11 11:57:48 Counter kolmogorov_smirnov 1.000000e+00 \n", + "16 2025-11-11 11:58:18 Counter kolmogorov_smirnov 1.000000e+00 \n", + "17 2025-11-11 11:58:48 Counter kolmogorov_smirnov 1.000000e+00 \n", + "18 2025-11-11 11:59:18 Counter kolmogorov_smirnov 1.000000e+00 \n", + "19 2025-11-11 11:59:48 Counter kolmogorov_smirnov 1.000000e+00 \n", + "30 2025-11-11 11:55:18 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "31 2025-11-11 11:55:48 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "32 2025-11-11 11:56:18 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "33 2025-11-11 11:56:48 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "34 2025-11-11 11:57:18 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "35 2025-11-11 11:57:48 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "36 2025-11-11 11:58:18 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "37 2025-11-11 11:58:48 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "38 2025-11-11 11:59:18 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "39 2025-11-11 11:59:48 Rollout kolmogorov_smirnov 1.000000e+00 \n", + "50 2025-11-11 11:55:18 multivariate multivariate 1.517720e-14 \n", + "51 2025-11-11 11:55:48 multivariate multivariate 1.588822e-14 \n", + "52 2025-11-11 11:56:18 multivariate multivariate 1.432145e-14 \n", + "53 2025-11-11 11:56:48 multivariate multivariate 1.421085e-14 \n", + "54 2025-11-11 11:57:18 multivariate multivariate 1.421085e-14 \n", + "55 2025-11-11 11:57:48 multivariate multivariate 1.776357e-15 \n", + "56 2025-11-11 11:58:18 multivariate multivariate 7.324107e-15 \n", + "57 2025-11-11 11:58:48 multivariate multivariate 1.464821e-14 \n", + "58 2025-11-11 11:59:18 multivariate multivariate 1.551137e-14 \n", + "59 2025-11-11 11:59:48 multivariate multivariate 1.427317e-14 \n", + "\n", + " p_value alert chunk_index chunk_start_date \\\n", + "10 None False 0 2025-11-11 11:55:18 \n", + "11 None False 1 2025-11-11 11:55:48 \n", + "12 None False 2 2025-11-11 11:56:18 \n", + "13 None False 3 2025-11-11 11:56:48 \n", + "14 None False 4 2025-11-11 11:57:18 \n", + "15 None False 5 2025-11-11 11:57:48 \n", + "16 None False 6 2025-11-11 11:58:18 \n", + "17 None False 7 2025-11-11 11:58:48 \n", + "18 None False 8 2025-11-11 11:59:18 \n", + "19 None False 9 2025-11-11 11:59:48 \n", + "30 None False 0 2025-11-11 11:55:18 \n", + "31 None False 1 2025-11-11 11:55:48 \n", + "32 None False 2 2025-11-11 11:56:18 \n", + "33 None False 3 2025-11-11 11:56:48 \n", + "34 None False 4 2025-11-11 11:57:18 \n", + "35 None False 5 2025-11-11 11:57:48 \n", + "36 None False 6 2025-11-11 11:58:18 \n", + "37 None False 7 2025-11-11 11:58:48 \n", + "38 None False 8 2025-11-11 11:59:18 \n", + "39 None False 9 2025-11-11 11:59:48 \n", + "50 NaN True 0 2025-11-11 11:55:18 \n", + "51 NaN True 1 2025-11-11 11:55:48 \n", + "52 NaN True 2 2025-11-11 11:56:18 \n", + "53 NaN True 3 2025-11-11 11:56:48 \n", + "54 NaN True 4 2025-11-11 11:57:18 \n", + "55 NaN True 5 2025-11-11 11:57:48 \n", + "56 NaN True 6 2025-11-11 11:58:18 \n", + "57 NaN True 7 2025-11-11 11:58:48 \n", + "58 NaN True 8 2025-11-11 11:59:18 \n", + "59 NaN True 9 2025-11-11 11:59:48 \n", + "\n", + " chunk_end_date \n", + "10 2025-11-11 11:55:18.999999999 \n", + "11 2025-11-11 11:55:48.999999999 \n", + "12 2025-11-11 11:56:18.999999999 \n", + "13 2025-11-11 11:56:48.999999999 \n", + "14 2025-11-11 11:57:18.999999999 \n", + "15 2025-11-11 11:57:48.999999999 \n", + "16 2025-11-11 11:58:18.999999999 \n", + "17 2025-11-11 11:58:48.999999999 \n", + "18 2025-11-11 11:59:18.999999999 \n", + "19 2025-11-11 11:59:48.999999999 \n", + "30 2025-11-11 11:55:18.999999999 \n", + "31 2025-11-11 11:55:48.999999999 \n", + "32 2025-11-11 11:56:18.999999999 \n", + "33 2025-11-11 11:56:48.999999999 \n", + "34 2025-11-11 11:57:18.999999999 \n", + "35 2025-11-11 11:57:48.999999999 \n", + "36 2025-11-11 11:58:18.999999999 \n", + "37 2025-11-11 11:58:48.999999999 \n", + "38 2025-11-11 11:59:18.999999999 \n", + "39 2025-11-11 11:59:48.999999999 \n", + "50 2025-11-11 11:55:18.999999999 \n", + "51 2025-11-11 11:55:48.999999999 \n", + "52 2025-11-11 11:56:18.999999999 \n", + "53 2025-11-11 11:56:48.999999999 \n", + "54 2025-11-11 11:57:18.999999999 \n", + "55 2025-11-11 11:57:48.999999999 \n", + "56 2025-11-11 11:58:18.999999999 \n", + "57 2025-11-11 11:58:48.999999999 \n", + "58 2025-11-11 11:59:18.999999999 \n", + "59 2025-11-11 11:59:48.999999999 " + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "# drop rows of drift with timestamp in reference data but not in analysis data\n", + "drift_filtered = drift[drift['timestamp'].isin(data['timestamp'])]\n", + "display(drift_filtered)\n", + "\n", + "drift_filtered.to_csv(\"drift_filtered.csv\", index=False)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "70671022", + "metadata": {}, + "outputs": [ + { + "ename": "ValueError", + "evalue": "Target column 'Square' not found in DataFrames", + "output_type": "error", + "traceback": [ + "\u001b[31m---------------------------------------------------------------------------\u001b[39m", + "\u001b[31mValueError\u001b[39m Traceback (most recent call last)", + "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[36]\u001b[39m\u001b[32m, line 6\u001b[39m\n\u001b[32m 3\u001b[39m \u001b[38;5;66;03m# rename column square to prediction\u001b[39;00m\n\u001b[32m 4\u001b[39m performance_reference.rename(columns={\u001b[33m'\u001b[39m\u001b[33mSquare\u001b[39m\u001b[33m'\u001b[39m: \u001b[33m'\u001b[39m\u001b[33mprediction\u001b[39m\u001b[33m'\u001b[39m}, inplace=\u001b[38;5;28;01mTrue\u001b[39;00m)\n\u001b[32m----> \u001b[39m\u001b[32m6\u001b[39m real_performance = \u001b[43mmodel\u001b[49m\u001b[43m.\u001b[49m\u001b[43mcalculate_performance\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 7\u001b[39m \u001b[43m \u001b[49m\u001b[43mreference_df\u001b[49m\u001b[43m=\u001b[49m\u001b[43mperformance_reference\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 8\u001b[39m \u001b[43m \u001b[49m\u001b[43manalysis_df\u001b[49m\u001b[43m=\u001b[49m\u001b[43mdata\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 9\u001b[39m \u001b[43m \u001b[49m\u001b[43mtarget_col\u001b[49m\u001b[43m=\u001b[49m\u001b[43mtarget_col\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 10\u001b[39m \u001b[43m \u001b[49m\u001b[43mtimestamp_col\u001b[49m\u001b[43m=\u001b[49m\u001b[43mtimestamp_col\u001b[49m\u001b[43m,\u001b[49m\n\u001b[32m 11\u001b[39m \u001b[43m \u001b[49m\u001b[43mchunk_period\u001b[49m\u001b[43m=\u001b[49m\u001b[43mchunk_period\u001b[49m\n\u001b[32m 12\u001b[39m \u001b[43m)\u001b[49m\n\u001b[32m 14\u001b[39m metrics_df = model.get_calculated_metrics_dataframe(real_performance)\n\u001b[32m 15\u001b[39m display(metrics_df)\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/sientia/ModelAnalysis.py:397\u001b[39m, in \u001b[36mModelAnalysis.calculate_performance\u001b[39m\u001b[34m(self, reference_df, analysis_df, target_col, timestamp_col, metrics, chunk_period)\u001b[39m\n\u001b[32m 395\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(DATAFRAMES_EMPTY_ERROR)\n\u001b[32m 396\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m target_col \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;129;01min\u001b[39;00m reference_df.columns \u001b[38;5;129;01mor\u001b[39;00m target_col \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;129;01min\u001b[39;00m analysis_df.columns:\n\u001b[32m--> \u001b[39m\u001b[32m397\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33mTarget column \u001b[39m\u001b[33m'\u001b[39m\u001b[38;5;132;01m{\u001b[39;00mtarget_col\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m'\u001b[39m\u001b[33m not found in DataFrames\u001b[39m\u001b[33m\"\u001b[39m)\n\u001b[32m 398\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m timestamp_col \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;129;01min\u001b[39;00m reference_df.columns \u001b[38;5;129;01mor\u001b[39;00m timestamp_col \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;129;01min\u001b[39;00m analysis_df.columns:\n\u001b[32m 399\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33mTimestamp column \u001b[39m\u001b[33m'\u001b[39m\u001b[38;5;132;01m{\u001b[39;00mtimestamp_col\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m'\u001b[39m\u001b[33m not found in DataFrames\u001b[39m\u001b[33m\"\u001b[39m)\n", + "\u001b[31mValueError\u001b[39m: Target column 'Square' not found in DataFrames" + ] + } + ], + "source": [ + "\n", + "real_performance = model.calculate_performance(\n", + " reference_df=data,\n", + " analysis_df=data,\n", + " target_col=target_col,\n", + " timestamp_col=timestamp_col,\n", + " chunk_period=chunk_period\n", + ")\n", + "\n", + "metrics_df = model.get_calculated_metrics_dataframe(real_performance)\n", + "display(metrics_df)\n", + "\n", + "metrics_df.to_csv(\"metrics_df.csv\", index=False)" + ] } ], "metadata": { diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 223227e..12f6901 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -44,8 +44,12 @@ metadata = { 'orchestrator.activities.formatters.minimal_retrain', return_value={'test_minimal_retrain': 'test_minimal_retrain'}, ) +@patch( + 'orchestrator.activities.formatters.drift', + return_value={'test_drift': 'test_drift'}, +) async def test_process_schedules( - mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters + mock_drift, mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters ): input_data = { 'pipelines': [ @@ -70,6 +74,13 @@ async def test_process_schedules( 'model_id': 'test_model_id', 'updated_at': '2021-01-03', }, + { + 'schedule_name': 'test_schedule_name4', + 'workflow_type': 'drift', + 'model_name': 'test_model_name', + 'model_id': 'test_model_id', + 'updated_at': '2021-01-04', + }, ] } @@ -88,11 +99,17 @@ async def test_process_schedules( 'test_minimal_retrain': 'test_minimal_retrain', 'updated_at': '2021-01-03', }, + 'test_schedule_name4': { + 'test_drift': 'test_drift', + 'updated_at': '2021-01-04', + }, }, } mock_scouter.assert_called_once_with(input_data['pipelines'][0]) mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1]) + mock_minimal_retrain.assert_called_once_with(input_data['pipelines'][2]) + mock_drift.assert_called_once_with(input_data['pipelines'][3]) @mark.asyncio diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index f74a936..7c8f5da 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -3,6 +3,7 @@ from unittest.mock import call, patch from orchestrator.utils.orchestrator_functions import ( build_tag_config, common_config, + drift, gather_read_tags, minimal_retrain, overlap_filter_config, @@ -35,6 +36,36 @@ def test_common_config(): assert result == expected +def test_drift(): + config = { + 'workflow_type': 'drift', + 'schedule_name': 'test_schedule', + 'model_id': 'test_model_id', + 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, + 'interval_minutes': 120, + 'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'], + } + result = drift(config) + expected = { + 'workflow_type': 'drift', + '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'}, + 'schema': 'sientia_data', + 'source_table_name': 'laborious_data', + 'target_table_name': 'drift_metrics', + 'interval': 120, + 'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'], + 'execution_timeout_seconds': 300, + 'task_timeout_seconds': 300, + } + assert result == expected + + def test_minimal_retrain(): config = { 'workflow_type': 'minimal_retrain', From 5cc7667fbc04464f1095d74891b74bb27a59f3d0 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 14 Nov 2025 12:20:36 -0300 Subject: [PATCH 2/8] SIENTIAPDE-1273 Update GITHUB_BRANCH to feature/SIENTIAPDE-1273 and increment Helm chart version to 0.6.0. Enhance TemporalManager to include offset in schedule intervals for improved scheduling flexibility. --- orchestrator/activities/temporal_manager.py | 6 ++++-- values.yaml | 4 ++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 6c423f5..5fd8197 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -320,8 +320,10 @@ class TemporalManager(SientiaMonitoring): input_data.description.schedule.spec.intervals = [ ScheduleIntervalSpec( - every=timedelta(seconds=parse_frequency( - schedule.get('frequency', '1m'))) + every=timedelta( + seconds=parse_frequency(schedule.get('frequency', '1m'))), + offset=timedelta( + seconds=parse_frequency(schedule.get('offset', '0m'))), ) ] diff --git a/values.yaml b/values.yaml index 88f5342..47e24af 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-1325-adicionar-metricas-especificas-de-operacoes-externas" + value: "feature/SIENTIAPDE-1273" - name: PYTHON_APP value: "orchestrator.worker.worker" @@ -237,7 +237,7 @@ ssh: # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp -# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.5.0 +# helm upgrade --install sientia-orchestrator-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.0 # kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \ # --namespace sientia \ From 7b28424a348cd7b283697d64fac215d2ba8af358 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 14 Nov 2025 15:55:06 -0300 Subject: [PATCH 3/8] SIENTIAPDE-1273 Implement simple metrics configuration in formatters and orchestrator utilities. Update tests to validate new simple metrics functionality and ensure proper integration with existing workflows. --- orchestrator/activities/formatters.py | 8 +++++ orchestrator/utils/orchestrator_functions.py | 15 ++++++++- .../activities/test_formatters.py | 23 +++++++++++++- .../activities/test_temporal_manager.py | 3 +- .../utils/test_orchestrator_functions.py | 31 +++++++++++++++++++ 5 files changed, 76 insertions(+), 4 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 73f8f86..b09df5b 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -20,6 +20,7 @@ with workflow.unsafe.imports_passed_through(): predictions_batch, scouter, drift, + simple_metrics, ) topic_separator = '\n ========== \n' @@ -140,6 +141,13 @@ class Formatters(SientiaMonitoring): '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) + ), + } 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 54c388b..65f1cc6 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -36,7 +36,6 @@ def drift(config: dict[str, Any]): """ Build drift configuration from pipeline config. """ - model = config['model'] return { **common_config(config), 'schema': 'sientia_data', @@ -51,6 +50,20 @@ def drift(config: dict[str, Any]): } +def simple_metrics(config: dict[str, Any]): + """ + Build simple metrics configuration from pipeline config. + """ + return { + **common_config(config), + 'schema': 'sientia_data', + 'predictions_table_name': 'predictions', + 'data_table_name': 'laborious_data', + 'interval_minutes': config.get('interval_minutes', 60), + 'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']), + } + + def minimal_retrain(config: dict[str, Any]): """ Build minimal retrain configuration from pipeline config. diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 12f6901..51b0e08 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -48,8 +48,17 @@ metadata = { '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_drift, mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters + mock_simple_metrics, + mock_drift, + mock_minimal_retrain, + mock_predictions_batch, + mock_scouter, + formatters, ): input_data = { 'pipelines': [ @@ -81,6 +90,13 @@ async def test_process_schedules( 'model_id': 'test_model_id', 'updated_at': '2021-01-04', }, + { + 'schedule_name': 'test_schedule_name5', + 'workflow_type': 'simple_metrics', + 'model_name': 'test_model_name', + 'model_id': 'test_model_id', + 'updated_at': '2021-01-05', + }, ] } @@ -103,6 +119,10 @@ async def test_process_schedules( 'test_drift': 'test_drift', 'updated_at': '2021-01-04', }, + 'test_schedule_name5': { + 'test_simple_metrics': 'test_simple_metrics', + 'updated_at': '2021-01-05', + }, }, } @@ -110,6 +130,7 @@ async def test_process_schedules( mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1]) mock_minimal_retrain.assert_called_once_with(input_data['pipelines'][2]) mock_drift.assert_called_once_with(input_data['pipelines'][3]) + mock_simple_metrics.assert_called_once_with(input_data['pipelines'][4]) @mark.asyncio diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index e3add38..f5a7c13 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -18,8 +18,7 @@ metadata = { @fixture -@patch('orchestrator.activities.temporal_manager.Client.connect') -def temporal_manager(connect_mock): +def temporal_manager(): temporal_manager = TemporalManager( host='localhost:7233', scouter_namespace='scouter', diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 7c8f5da..de0eddd 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -10,6 +10,7 @@ from orchestrator.utils.orchestrator_functions import ( predictions_batch, process_path_priority, scouter, + simple_metrics, ) @@ -66,6 +67,36 @@ def test_drift(): assert result == expected +def test_simple_metrics(): + config = { + 'workflow_type': 'simple_metrics', + 'schedule_name': 'test_schedule', + 'model_id': 'test_model_id', + 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, + 'interval_minutes': 120, + 'metrics': ['rmse', 'mse'], + } + result = simple_metrics(config) + expected = { + 'workflow_type': 'simple_metrics', + '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'}, + 'schema': 'sientia_data', + 'predictions_table_name': 'predictions', + 'data_table_name': 'laborious_data', + 'interval_minutes': 120, + 'metrics': ['rmse', 'mse'], + 'execution_timeout_seconds': 300, + 'task_timeout_seconds': 300, + } + assert result == expected + + def test_minimal_retrain(): config = { 'workflow_type': 'minimal_retrain', From 7d0c23acb613985b75bbbd5b40c3aebe90b39941 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 17 Nov 2025 10:47:40 -0300 Subject: [PATCH 4/8] SIENTIAPDE-1273 Update sientia-dataops-library dependency to version 1.6.1 and add target_table_name to simple_metrics configuration for enhanced metrics tracking. --- orchestrator/utils/orchestrator_functions.py | 1 + requirements.txt | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 65f1cc6..d3f4a06 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -59,6 +59,7 @@ def simple_metrics(config: dict[str, Any]): 'schema': 'sientia_data', 'predictions_table_name': 'predictions', 'data_table_name': 'laborious_data', + 'target_table_name': 'simple_metrics', 'interval_minutes': config.get('interval_minutes', 60), 'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']), } diff --git a/requirements.txt b/requirements.txt index 4c94dc5..cadaefb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,5 +4,5 @@ sqlalchemy redis pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.6.1 prometheus-client From 2f06036d0e6ef1db5fbc7b4ce3a853ddd96d38ca Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 17 Nov 2025 12:57:17 -0300 Subject: [PATCH 5/8] SIENTIAPDE-1273 SIENTIAPDE-1273 Refactor drift function in orchestrator utilities for improved configuration and update formatters to include drift import. Enhance tests to validate new configurations and ensure integration with existing workflows. --- orchestrator/activities/formatters.py | 2 +- orchestrator/utils/orchestrator_functions.py | 8 +++----- tests/orchestrator/utils/test_orchestrator_functions.py | 1 + 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index b09df5b..6123ad2 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -15,11 +15,11 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.utils.orchestrator_functions import ( build_tag_config, + drift, gather_read_tags, minimal_retrain, predictions_batch, scouter, - drift, simple_metrics, ) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index d3f4a06..3aec5dd 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -42,11 +42,9 @@ def drift(config: dict[str, Any]): 'source_table_name': 'laborious_data', 'target_table_name': 'drift_metrics', 'interval': config.get('interval_minutes', 60), - 'drift_metrics': config.get('drift_metrics', [ - "kolmogorov_smirnov", - "jensen_shannon", - "wasserstein" - ]) + 'drift_metrics': config.get( + 'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'] + ), } diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index de0eddd..4dc9666 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -89,6 +89,7 @@ def test_simple_metrics(): 'schema': 'sientia_data', 'predictions_table_name': 'predictions', 'data_table_name': 'laborious_data', + 'target_table_name': 'simple_metrics', 'interval_minutes': 120, 'metrics': ['rmse', 'mse'], 'execution_timeout_seconds': 300, From fdd859de5a9bb57be862c31545281d900e3c23fa Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 17 Nov 2025 16:30:29 -0300 Subject: [PATCH 6/8] SIENTIAPDE-1273 Enhance orchestrator utilities and activities with improved documentation and functionality. Update README to reflect new orchestrator functions for drift and simple metrics workflows. Refactor email, formatters, mongo_db, slot_manager, and temporal_manager activities to include detailed docstrings and improve clarity on input parameters and return values. Ensure consistent metadata handling across activities for better logging and tracking. --- README.md | 2 +- orchestrator/activities/email.py | 33 ++-- orchestrator/activities/formatters.py | 170 ++++++++++++------- orchestrator/activities/mongo_db.py | 42 +++-- orchestrator/activities/slot_manager.py | 49 ++++-- orchestrator/activities/temporal_manager.py | 9 +- orchestrator/utils/email_builder.py | 28 ++- orchestrator/utils/orchestrator_functions.py | 51 +++++- orchestrator/worker/worker.py | 9 +- 9 files changed, 271 insertions(+), 122 deletions(-) diff --git a/README.md b/README.md index 134578b..645679f 100644 --- a/README.md +++ b/README.md @@ -412,7 +412,7 @@ The orchestrator includes an advanced notification filtering system that prevent #### **Utilities (`orchestrator/utils/`)** - **Connectors Configuration**: Database and service configuration management - **Email Builder**: HTML email template generation and formatting using Jinja2 templates -- **Orchestrator Functions**: Pipeline configuration transformation utilities for scouter, predictions_batch, and minimal_retrain workflows +- **Orchestrator Functions**: Pipeline configuration transformation utilities for scouter, predictions_batch, minimal_retrain, drift, and simple_metrics workflows - **Converters**: Data type conversion and validation utilities including frequency parsing - **Templates**: HTML email templates for alerts and reports diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py index bbad1a0..ab30c78 100644 --- a/orchestrator/activities/email.py +++ b/orchestrator/activities/email.py @@ -71,6 +71,8 @@ class Email(SientiaMonitoring): def close(self): """ Close the Email connection and clean up resources. + + Closes the SMTP server connection and shuts down the SientiaMonitoring instance. """ self.server.quit() SientiaMonitoring.shutdown(self) @@ -120,15 +122,23 @@ class Email(SientiaMonitoring): def handle_attachments(self, attachments: list[dict], msg: MIMEMultipart) -> MIMEMultipart: """ - Attaches a list of attachments to an email message. + Attach a list of attachments to an email message. + + Processes notification attachments and adds them to the email message + as base64-encoded MIME parts. Each attachment contains error details + or additional context for the notification. + Args: - attachments (List[Dict]): A list of dictionaries where each dictionary contains - the keys 'filename' and 'content' representing the attachment details. - msg (MIMEMultipart): The email message object to which the attachments will be added. + attachments (list[dict]): A list of dictionaries where each dictionary contains: + - filename (str): Name of the attachment file + - attachment_content (str): Content of the attachment + msg (MIMEMultipart): The email message object to which the attachments will be added + Returns: - MIMEMultipart: The email message object with the attachments added. + MIMEMultipart: The email message object with the attachments added + Raises: - Exception: If an attachment cannot be added, an error is logged. + Exception: If an attachment cannot be added, an error is logged and raised """ for attachment in attachments: @@ -149,14 +159,17 @@ class Email(SientiaMonitoring): def try_send_email(self, msg: MIMEMultipart, receivers: str): """ - Sends an email to the receivers with automatic reconnection handling. + Send an email to the receivers with automatic reconnection handling. + + Attempts to send an email and automatically reconnects to the SMTP server + if a disconnection occurs during transmission. Args: - msg (MIMEMultipart): The email message to send. - receivers (str): Comma-separated list of email addresses to send to. + msg (MIMEMultipart): The email message to send + receivers (str): Comma-separated list of email addresses to send to Raises: - Exception: If email sending fails after reconnection attempts. + Exception: If email sending fails after reconnection attempts """ try: self.server.sendmail(self.sender_email, receivers, msg.as_string()) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 6123ad2..9b7ce92 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -69,6 +69,8 @@ class Formatters(SientiaMonitoring): def close(self): """ Close the Formatters connection and clean up resources. + + Shuts down the SientiaMonitoring instance and releases all resources. """ SientiaMonitoring.shutdown(self) @@ -229,14 +231,18 @@ class Formatters(SientiaMonitoring): @activity.defn(name='format_schedule_config') async def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Formats the schedule config to a dictionary with the schedule name as the key. + Format the schedule config to a dictionary with the schedule name as the key. + + Transforms a list of schedule configurations into a nested dictionary structure + organized by namespace and schedule name for efficient lookup and comparison. Args: - input_data (dict[str, Any]): The input data containing the schedule config to format. - - schedule_config (list[dict[str, Any]]): The schedule config to format. + input_data (dict[str, Any]): The input data containing: + - schedule_config (list[dict[str, Any]]): The schedule config to format + - metadata (dict): Metadata for logging purposes Returns: - dict[str, Any]: The formatted schedule config. + dict[str, Any]: The formatted schedule config organized by namespace and schedule name """ metadata = input_data['metadata'] @@ -272,13 +278,16 @@ class Formatters(SientiaMonitoring): """ Compare new and current schedules to determine which should be updated or created. + Compares schedule timestamps to identify schedules that need updating (newer timestamp) + or creating (schedule doesn't exist). Results are accumulated in the provided dictionaries. + Args: - schedules (dict[str, Any]): New schedules to compare. - current_schedules (dict[str, Any]): Existing schedules to compare against. - to_update (dict[str, Any]): Output accumulator for schedules that need updating. - to_create (dict[str, Any]): Output accumulator for schedules that need creating. - namespace (str): Namespace for the schedules being compared. - metadata (dict[str, Any]): Metadata for logging purposes. + schedules (dict[str, Any]): New schedules to compare + current_schedules (dict[str, Any]): Existing schedules to compare against + to_update (dict[str, Any]): Output accumulator for schedules that need updating + to_create (dict[str, Any]): Output accumulator for schedules that need creating + namespace (str): Namespace for the schedules being compared + metadata (dict[str, Any]): Metadata for logging purposes """ for schedule_name, schedule in schedules.items(): if schedule_name in current_schedules: @@ -299,19 +308,23 @@ class Formatters(SientiaMonitoring): @activity.defn(name='create_schedule_config') async def create_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Creates a schedule config dictionary based on the input data. - Checks the existing schedule config and updates it with the new schedule config, - deleting unnecessary schedules, creating new schedules and updating existing schedules. + Create a schedule config dictionary based on the input data. + + Compares the current schedule configuration in Temporal with the new schedule + configuration to determine which schedules need to be created, updated, or deleted. + Uses timestamp comparison to identify schedules that have changed. Args: - - input_data (dict[str, Any]): The input data containing - the schedules to process. - - current_schedule_config (dict[str, Any]): The current schedule - config in Temporal server. - - schedule_config (dict[str, Any]): The schedule config to process. + input_data (dict[str, Any]): The input data containing: + - current_schedule_config (dict[str, Any]): The current schedule config in Temporal server + - schedule_config (dict[str, Any]): The schedule config to process + - metadata (dict): Metadata for logging purposes Returns: - - dict[str, Any]: A dictionary with keys 'to_update', 'to_create', and 'to_delete' + dict[str, Any]: A dictionary with keys: + - to_update (dict): Schedules that need updating + - to_create (dict): Schedules that need creating + - to_delete (dict): Schedules that need deleting """ metadata = input_data.get('metadata', {}) @@ -355,19 +368,22 @@ class Formatters(SientiaMonitoring): @activity.defn(name='create_slot_config') async def create_slot_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Creates a slot config dictionary based on the input data. - Checks the existing slot config and updates it with the new slot config, - deleting unnecessary slots. + Create a slot config dictionary based on the input data. + + Compares the current slot configuration in Redis with the new slot configuration + to determine which slots need to be inserted or deleted. Slots are identified + by numeric IDs, and excess slots are marked for deletion. Args: - - input_data (dict[str, Any]): The input data containing - the slots to process. - - current_slot_config (dict[str, Any]): The current slot - config in Temporal server. - - slot_config (dict[str, Any]): The slot config to process. + input_data (dict[str, Any]): The input data containing: + - current_slot_config (dict[str, Any]): The current slot config in Redis + - slot_config (dict[str, Any]): The slot config to process + - metadata (dict): Metadata for logging purposes Returns: - - dict[str, Any]: A dictionary containing 'to_insert' and 'to_delete' + dict[str, Any]: A dictionary containing: + - to_insert (dict): Slots that need to be inserted/updated + - to_delete (list[str]): Slot IDs that need to be deleted """ metadata = input_data.get('metadata', {}) @@ -399,13 +415,16 @@ class Formatters(SientiaMonitoring): attachment: Any | None = None, ) -> None: """ - Sends a success notification report. + Send a success notification report. + + Sends an INFO-level notification to the notification handler with success + details about orchestration operations. Args: - metadata (dict[str, Any]): Metadata for the notification. - message (str): The success message to send. - notification_id (str): The ID of the notification to send. - attachment (Any | None, optional): Optional attachment content to include. + metadata (dict[str, Any]): Metadata for the notification + message (str): The success message to send + notification_id (str): The ID of the notification to send + attachment (Any | None, optional): Optional attachment content to include """ await self.send_notification_async( metadata=metadata, @@ -420,13 +439,16 @@ class Formatters(SientiaMonitoring): self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str ) -> None: """ - Sends an error notification report. + Send an error notification report. + + Sends an ERROR-level notification to the notification handler with error + details about orchestration operation failures. Args: - metadata (dict[str, Any]): Metadata for the notification. - message (str): The error message to send. - notification_id (str): The ID of the notification. - attachment (str): The attachment content for the notification. + metadata (dict[str, Any]): Metadata for the notification + message (str): The error message to send + notification_id (str): The ID of the notification + attachment (str): The attachment content for the notification """ await self.send_notification_async( metadata=metadata, @@ -441,12 +463,15 @@ class Formatters(SientiaMonitoring): self, input_data: list[dict[str, Any]] ) -> tuple[list[str], dict[str, Any]]: """ - Parses the report schedule data to extract success and error information. + Parse the report schedule data to extract success and error information. + + Processes schedule operation reports and separates successful operations + from failed ones, formatting keys as "namespace/schedule_name" for consistency. Args: input_data (list[dict[str, Any]]): The schedule reports. Each item must contain 'namespace', 'schedule_name', 'success', 'message', and optionally - 'attachment'. + 'attachment' Returns: tuple[list[str], dict[str, Any]]: A tuple of: @@ -472,11 +497,14 @@ class Formatters(SientiaMonitoring): def parse_report(self, input_data: dict[str, dict[str, Any]]) -> tuple[list[str], list[str]]: """ - Parses the report data to extract success and error keys. + Parse the report data to extract success and error keys. + + Processes operation reports and separates successful operations from failed ones + based on the 'success' field in each report item. Args: - input_data (dict[str, Any]): The input data containing report items. - Each item should have a 'success' field indicating success/failure. + input_data (dict[str, dict[str, Any]]): The input data containing report items. + Each item should have a 'success' field indicating success/failure Returns: tuple[list[str], list[str]]: A tuple containing: @@ -498,14 +526,17 @@ class Formatters(SientiaMonitoring): schedule_data: dict[str, Any], ): """ - Manages and sends success and error reports based on the provided keys and data. + Manage and send success and error reports based on the provided keys and data. + + Sends separate notifications for successful and failed operations, formatting + error messages with attachments when available. Args: - metadata (dict[str, Any]): Metadata for logging and notifications. - success_keys (list[str]): List of keys that were successful. - error_keys (dict[str, Any]): Dictionary of error keys mapped to error details. - schedule_type (str): The type of schedule being reported (e.g., 'created schedules'). - schedule_data (dict[str, Any]): The schedule data containing items and notification ID. + metadata (dict[str, Any]): Metadata for logging and notifications + success_keys (list[str]): List of keys that were successful + error_keys (dict[str, Any]): Dictionary of error keys mapped to error details + schedule_type (str): The type of schedule being reported (e.g., 'created schedules') + schedule_data (dict[str, Any]): The schedule data containing items and notification ID """ if len(success_keys) > 0: await self.send_success_report( @@ -533,14 +564,17 @@ class Formatters(SientiaMonitoring): @activity.defn(name='report_schedule_orchestration') async def report_schedule_orchestration(self, input_data: dict[str, Any]) -> None: """ - Reports the orchestration result to the notification handler. + Report the orchestration result to the notification handler. + + Processes schedule orchestration results and sends notifications for + created, updated, and deleted schedules with success and error details. Args: - - input_data (dict[str, Any]): The input data containing - the orchestration result. - - created_schedules (dict[str, Any]): The created schedules. - - updated_schedules (dict[str, Any]): The updated schedules. - - deleted_schedules (list[str]): The deleted schedules. + input_data (dict[str, Any]): The input data containing: + - created_schedules (list[dict[str, Any]]): The created schedules + - updated_schedules (list[dict[str, Any]]): The updated schedules + - deleted_schedules (list[dict[str, Any]]): The deleted schedules + - metadata (dict): Metadata for logging purposes """ metadata = input_data.get('metadata', {}) @@ -580,13 +614,16 @@ class Formatters(SientiaMonitoring): @activity.defn(name='report_slot_orchestration') async def report_slot_orchestration(self, input_data: dict[str, Any]) -> None: """ - Reports the orchestration result to the notification handler. + Report the slot orchestration result to the notification handler. + + Processes slot orchestration results and sends notifications for + inserted and deleted slots with success and error details. Args: - - input_data (dict[str, Any]): The input data containing - the orchestration result. - - inserted_slots (dict[str, Any]): The inserted slots. - - deleted_slots (list[str]): The deleted slots. + input_data (dict[str, Any]): The input data containing: + - inserted_slots (dict[str, Any]): The inserted slots + - deleted_slots (dict[str, Any]): The deleted slots + - metadata (dict): Metadata for logging purposes """ metadata = input_data.get('metadata', {}) @@ -635,16 +672,19 @@ class Formatters(SientiaMonitoring): @activity.defn(name='format_log_report') async def format_log_report(self, input_data: dict[str, Any]) -> dict[Hashable, Any]: """ - Formats the receiver_groups status to a dataframe to be stored in the database. + Format the receiver_groups status to a dataframe to be stored in the database. + + Transforms receiver group notification data into a structured format suitable + for database storage, aggregating notifications by unique notification ID and trigger. Args: input_data (dict[str, Any]): The input data containing: - - receiver_groups (dict): The receiver groups configuration. - - mail_type (str): The type of mail for the report. - - metadata (dict): Metadata for logging purposes. + - receiver_groups (dict): The receiver groups configuration with notifications + - mail_type (str): The type of mail for the report (Alerts/Reports) + - metadata (dict): Metadata for logging purposes Returns: - dict[str, Any]: The formatted log report as a dictionary representation of a DataFrame. + dict[Hashable, Any]: The formatted log report as a dictionary representation of a DataFrame """ metadata = input_data['metadata'] mail_type = input_data['mail_type'] diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 16dbb2d..0339ab5 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -69,6 +69,8 @@ class MongoDB(SientiaMonitoring): def close(self): """ Shutdown the MongoDB client and clean up resources. + + Closes the MongoDB repository connection and shuts down the SientiaMonitoring instance. """ self.mongo_db_repository.close() SientiaMonitoring.shutdown(self) @@ -220,8 +222,13 @@ class MongoDB(SientiaMonitoring): """ Update `updated_at` timestamps for successfully updated pipelines. - input_data: - - updated_pipelines (list[dict]): Pipelines with success flags to consider. + Updates the `updated_at` field in the `orchestrated_schedules` collection + for all pipelines that were successfully updated in Temporal. + + Args: + input_data (dict[str, Any]): Input data containing: + - updated_pipelines (list[dict]): Pipelines with success flags to consider + - metadata (dict): Metadata for logging purposes """ updated_pipelines = input_data.get('updated_pipelines', []) metadata = input_data.get('metadata', {}) @@ -263,8 +270,13 @@ class MongoDB(SientiaMonitoring): """ Insert `updated_at` timestamps for newly created pipelines. - input_data: - - created_pipelines (list[dict]): Pipelines with success flags to consider. + Inserts new documents into the `orchestrated_schedules` collection + for all pipelines that were successfully created in Temporal. + + Args: + input_data (dict[str, Any]): Input data containing: + - created_pipelines (list[dict]): Pipelines with success flags to consider + - metadata (dict): Metadata for logging purposes """ created_pipelines = input_data.get('created_pipelines', []) metadata = input_data.get('metadata', {}) @@ -312,8 +324,13 @@ class MongoDB(SientiaMonitoring): """ Delete timestamp rows for successfully deleted pipelines. - input_data: - - deleted_pipelines (list[dict]): Pipelines with success flags to consider. + Removes documents from the `orchestrated_schedules` collection + for all pipelines that were successfully deleted from Temporal. + + Args: + input_data (dict[str, Any]): Input data containing: + - deleted_pipelines (list[dict]): Pipelines with success flags to consider + - metadata (dict): Metadata for logging purposes """ deleted_pipelines = input_data.get('deleted_pipelines', []) metadata = input_data.get('metadata', {}) @@ -352,10 +369,15 @@ class MongoDB(SientiaMonitoring): @activity.defn(name='create_collection_with_ttl_index') async def create_collection_with_ttl_index(self, input_data: dict[str, Any]) -> None: """ - Create a collection with a TTL index. - input_data: - - collection_name (str): The name of the collection to create. - - ttl_index (str): The name of the TTL index to create. + Create collections with TTL indexes for pipeline topics. + + Creates MongoDB collections for scouter pipeline topics and sets up + TTL indexes on the `inserted_at` field to automatically expire old documents. + + Args: + input_data (dict[str, Any]): Input data containing: + - pipelines (dict[str, Any]): Pipeline configurations with topic names + - metadata (dict): Metadata for logging purposes """ pipelines = input_data.get('pipelines', {}) metadata = input_data.get('metadata', {}) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 953c3a6..26b6813 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -71,6 +71,8 @@ class SlotManager(SientiaMonitoring): def close(self): """ Close the SlotManager connection and clean up resources. + + Closes the Redis repository connection and shuts down the SientiaMonitoring instance. """ self.redis_repository.close() SientiaMonitoring.shutdown(self) @@ -142,10 +144,16 @@ class SlotManager(SientiaMonitoring): @activity.defn(name='load_active_ingestors') async def load_active_ingestors(self, input_data: dict[str, Any]) -> list[str]: """ - Load all active ingestors from Redis + Load all active ingestors from Redis. + + Retrieves all active ingestor heartbeat keys from Redis to determine + which ingestors are currently available for slot assignment. + + Args: + input_data (dict[str, Any]): Activity input containing metadata Returns: - list[str]: A list of active ingestors + list[str]: A list of active ingestor keys from Redis """ metadata = input_data.get('metadata', {}) @@ -259,15 +267,19 @@ class SlotManager(SientiaMonitoring): @activity.defn(name='get_last_data_timestamp') async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None: """ - Gets the last data timestamp from redis. + Get the last data timestamp from Redis. + + Retrieves the last processed timestamp for a specific mail type from Redis. + This timestamp is used for incremental data loading to avoid reprocessing + already processed notifications. Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - mail_type (str): The type of mail to get timestamp for. + - metadata (dict): Metadata for logging purposes + - mail_type (str): The type of mail to get timestamp for (Alerts/Reports) Returns: - str | None: The last data timestamp as a string, or None if no timestamp exists. + str | None: The last data timestamp as a string, or None if no timestamp exists """ metadata = input_data['metadata'] key = f'notification_last_timestamp:{input_data["mail_type"]}' @@ -293,18 +305,22 @@ class SlotManager(SientiaMonitoring): return data_hold @activity.defn(name='put_last_data_timestamp') - async def put_last_data_timestamp(self, input_data: dict[str, Any]): + async def put_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None: """ - Puts the last data timestamp into redis. + Store the last data timestamp in Redis. + + Extracts the maximum timestamp from the provided data and stores it in Redis + with a TTL for the specified mail type. This enables incremental processing + in subsequent workflow executions. Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - data (list[dict]): The data to extract timestamp from. - - mail_type (str): The type of mail to store timestamp for. + - metadata (dict): Metadata for logging purposes + - data (list[dict]): The data to extract timestamp from + - mail_type (str): The type of mail to store timestamp for (Alerts/Reports) Returns: - str | None: The last data timestamp that was stored, or None if no data exists. + str | None: The last data timestamp that was stored, or None if no data exists """ metadata = input_data['metadata'] key = f'notification_last_timestamp:{input_data["mail_type"]}' @@ -426,11 +442,14 @@ class SlotManager(SientiaMonitoring): """ Store notification cache in Redis to track recently sent notifications. + Stores successfully sent notifications in Redis with a TTL to prevent + duplicate alert delivery. Only notifications with status 'sent' are cached. + Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - log_report (list[dict]): The log report containing notification statuses. - - sent_ttl (int): Time to live for sent notification cache in seconds. + - metadata (dict): Metadata for logging purposes + - log_report (list[dict]): The log report containing notification statuses + - sent_ttl (int): Time to live for sent notification cache in seconds """ metadata = input_data['metadata'] log_report = DataFrame(input_data['log_report']) diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 5fd8197..6b0932a 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -68,6 +68,8 @@ class TemporalManager(SientiaMonitoring): def close(self): """ Close the TemporalManager connection and clean up resources. + + Shuts down the SientiaMonitoring instance and releases all resources. """ SientiaMonitoring.shutdown(self) @@ -80,7 +82,9 @@ class TemporalManager(SientiaMonitoring): async def connect_to_temporal(self): """ Connect to Temporal server namespaces used by scouter and laborious workflows. + Creates and caches `Client` connections for both namespaces for later use. + The connections are stored in `temporal_clients` dictionary for efficient access. """ self.logger.info(f'Connecting to Temporal side namespaces at {self.temporal_host}') self.logger.info(f'Scouter namespace: {self.scouter_namespace}') @@ -108,8 +112,9 @@ class TemporalManager(SientiaMonitoring): Any schedule marked with search attribute `orchestrated=true` that does not exist in MongoDB collection `orchestrated_schedules` will be deleted. - input_data: - - orchestrated_schedules (dict[str, Any]): Current orchestrated schedules from MongoDB. + Args: + input_data (dict[str, Any]): Input data containing: + - orchestrated_schedules (dict[str, Any]): Current orchestrated schedules from MongoDB. """ metadata = input_data['metadata'] diff --git a/orchestrator/utils/email_builder.py b/orchestrator/utils/email_builder.py index db67591..c6e4d23 100644 --- a/orchestrator/utils/email_builder.py +++ b/orchestrator/utils/email_builder.py @@ -32,12 +32,15 @@ class EmailBuilder: """ Replace parameters in a Jinja2 template with provided values. + Renders a Jinja2 template string with the provided parameter dictionary, + replacing all template variables with their corresponding values. + Args: - template (str): The Jinja2 template string. - parameters (dict): Dictionary of parameters to replace in the template. + template (str): The Jinja2 template string + parameters (dict): Dictionary of parameters to replace in the template Returns: - str: The rendered template with parameters replaced. + str: The rendered template with parameters replaced """ # Create a Jinja2 template from the provided string template_obj = Template(template) @@ -48,13 +51,16 @@ class EmailBuilder: """ Build parameters dictionary for email templates based on general events and mail type. + Processes notification events organized by level and model, rendering HTML + sections for each notification level using the general template. + Args: general_events (dict): Dictionary containing events categorized by level (ERROR, WARNING, INFO). - Each level contains a 'models' key with model-specific event data. - mail_type (str): The type of email being sent. + Each level contains a 'models' key with model-specific event data + mail_type (str): The type of email being sent (Alerts/Reports) Returns: - dict: Dictionary with mail_type and rendered event sections for each notification level. + dict: Dictionary with mail_type and rendered event sections for each notification level """ error_events = general_events.get('ERROR', {}) warning_events = general_events.get('WARNING', {}) @@ -79,16 +85,20 @@ class EmailBuilder: def build_email(self, report_data: list[dict[str, Any]], mail_type: str) -> str: """ - Builds the email HTML by organizing report data by notification level and model. + Build the email HTML by organizing report data by notification level and model. + + Organizes notification data by level and model, then renders the complete + HTML email using the report template with all event sections. Args: - report_data (List[Dict[str, Any]]): List of notification reports, each containing: + report_data (list[dict[str, Any]]): List of notification reports, each containing: - level (str): Notification level (ERROR, WARNING, INFO) - model_name (str): Name of the model - Additional notification details + mail_type (str): The type of email being built (Alerts/Reports) Returns: - str: Complete HTML email content ready for sending. + str: Complete HTML email content ready for sending """ general_events: dict[str, dict[str, Any]] = {} diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 3aec5dd..ac50f32 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -35,6 +35,19 @@ def common_config(config: dict[str, Any]): def drift(config: dict[str, Any]): """ Build drift configuration from pipeline config. + + Creates a drift detection workflow configuration with database table mappings + and drift metric specifications for monitoring data distribution changes. + + 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']) + - Additional fields from common_config + + Returns: + dict[str, Any]: Drift configuration with workflow type set to 'drift' """ return { **common_config(config), @@ -51,6 +64,19 @@ def drift(config: dict[str, Any]): def simple_metrics(config: dict[str, Any]): """ Build simple metrics configuration from pipeline config. + + Creates a simple metrics computation workflow configuration for calculating + model performance metrics like RMSE, MSE, MAE, and R². + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - interval_minutes (int, optional): Computation interval in minutes (default: 60) + - metrics (list[str], optional): List of metrics to compute + (default: ['rmse', 'mse', 'mae', 'r2']) + - Additional fields from common_config + + Returns: + dict[str, Any]: Simple metrics configuration with workflow type set to 'simple_metrics' """ return { **common_config(config), @@ -245,13 +271,17 @@ def predictions_batch(config: dict[str, Any]): def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: """ - Gathers all read tags from input pipelines. + Gather all read tags from input pipelines. + + Collects all read tags from scouter pipelines and organizes them by + server_id and tag_address, tracking which topics each tag is associated with. Args: - - pipelines (list[dict[str, Any]]): The schedules to process. + pipelines (list[dict[str, Any]]): The pipeline configurations to process Returns: - - dict[str, Any]: The read tags dictionary + dict[str, Any]: Dictionary of read tags keyed by "server_id:tag_address", + each containing tag configuration and associated topics """ tags = {} @@ -274,19 +304,24 @@ def build_tag_config( """ Build tag configuration for a specific slot and OPC server. + Organizes tags by OPC server and calculates the minimum subscription period + based on tag frequencies. Validates that all server IDs exist in the OPC + servers configuration. + Args: tags (list[dict[str, Any]]): List of tag configurations containing: - server_id (str): ID of the OPC server - tag_address (str): Address of the tag - slot_config (dict[str, Any]): Current slot configuration to update. - opc_servers (dict[str, Any]): Dictionary of OPC server configurations. - i (int): Slot number to configure. + - frequency (int): Tag read frequency in milliseconds + opc_servers (dict[str, Any]): Dictionary of OPC server configurations Returns: - dict[str, Any]: Updated slot configuration with the new tag. + tuple[dict[str, Any], list]: A tuple containing: + - Slot configuration dictionary organized by server name + - List of server IDs that were not found in opc_servers Raises: - ValueError: If the specified server_id is not found in opc_servers. + ValueError: If the specified server_id is not found in opc_servers """ slot_config = {} diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 6283cbd..fe65d4f 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -39,7 +39,8 @@ async def main(): Sets up MongoDB connection, notification handler, Temporal client, and starts multiple workers for different task queues (orchestrator, alerts, reports). - Handles graceful shutdown and error handling. + Handles graceful shutdown and error handling. Initializes Prometheus metrics + server and SDK metrics for monitoring. """ host = os.getenv('TEMPORAL_HOST', 'localhost:7233') namespace = os.getenv('TEMPORAL_NAMESPACE', 'default') @@ -206,7 +207,11 @@ def start_prometheus_server(): Start the Prometheus metrics server on the configured port. Sets up HTTP server for metrics collection and marks the application as UP. - Exits the application if the server fails to start. + Exits the application if the server fails to start. The metrics server + exposes application metrics on the port specified by HTTP_METRICS_PORT. + + Raises: + SystemExit: If the metrics server fails to start """ try: port = int(os.getenv('HTTP_METRICS_PORT', 9090)) From 81732d1409ab5575838aa78b26c9ac5e2dcb8ec7 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 18 Nov 2025 09:20:11 -0300 Subject: [PATCH 7/8] SIENTIAPDE-1273 Update image tag in values.yaml from 0.5.0 to 0.5.2 for version consistency. --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index 47e24af..a40c8c4 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.5.0" + tag: "0.5.2" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: From 62e7a7dee987904e5ce4367578a1b9dc1a890aee Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 19 Nov 2025 08:12:08 -0300 Subject: [PATCH 8/8] SIENTIAPDE-1273 SIENTIAPDE-1273 Add 'save_transform' and 'transform_table_name' parameters to predictions_batch function for enhanced data processing capabilities. Update corresponding test cases to reflect these changes. --- orchestrator/utils/orchestrator_functions.py | 2 ++ tests/orchestrator/utils/test_orchestrator_functions.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index ac50f32..b8e7eec 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -248,6 +248,8 @@ def predictions_batch(config: dict[str, Any]): 'datetime_columns': config.get('datetime_columns', []), 'schema': 'sientia_data', 'table_name': 'predictions', + 'save_transform': config.get('save_transform', True), + 'transform_table_name': 'transformed_data', 'retention_time': config.get('model_retention_minutes', 60) * 60, 'opc_output_config': tags, 'input_filters': overlap_filter_config( diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 4dc9666..4db42a8 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -238,6 +238,8 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi 'query': 'test_query', 'schema': 'sientia_data', 'table_name': 'predictions', + 'save_transform': True, + 'transform_table_name': 'transformed_data', 'retention_time': 60 * 60, 'opc_output_config': { 'test_server_id': {