From 64f0747e8e83887d88730dfec399d0a6bf670a17 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 14 Nov 2025 15:43:28 -0300 Subject: [PATCH] SIENTIAPDE-1273 Update version and enhance metrics calculation in Laborious system - Updated image tag in values.yaml from 1.1.0 to 1.1.1. - Modified GITHUB_BRANCH environment variable for consistency. - Added a new method `calculate_simple_metrics` in model_metrics.py to compute various model performance metrics including RMSE, MSE, MAE, and R2. - Integrated the new metrics calculation into the worker setup, allowing for concurrent processing of simple metrics. - Updated tests to cover the new metrics calculation functionality, ensuring comprehensive validation of the implementation. --- laborious/activities/model_metrics.py | 83 +++++- laborious/worker/worker.py | 18 ++ laborious/workflows/drift.py | 3 +- laborious/workflows/simple_metrics.py | 95 +++++++ .../activities/test_model_metrics.py | 252 ++++++++++++++++++ tests/laborious/workflows/test_drift.py | 18 +- .../workflows/test_simple_metrics.py | 228 ++++++++++++++++ values.yaml | 6 +- 8 files changed, 692 insertions(+), 11 deletions(-) create mode 100644 laborious/workflows/simple_metrics.py create mode 100644 tests/laborious/workflows/test_simple_metrics.py diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index 4e24ec7..c70cc81 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -13,6 +13,7 @@ with workflow.unsafe.imports_passed_through(): from sientia.ModelAnalysis import ModelAnalysis from laborious import metrics import time + import numpy as np from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ import warnings import traceback @@ -20,7 +21,6 @@ with workflow.unsafe.imports_passed_through(): warnings.filterwarnings('ignore', category=RuntimeWarning, message='Degrees of freedom <= 0') warnings.filterwarnings('ignore', category=RuntimeWarning, message='invalid value encountered in scalar divide') - class ModelMetrics(SientiaMonitoring): """ Metrics activities for the Laborious system. @@ -264,3 +264,84 @@ class ModelMetrics(SientiaMonitoring): return drift_df.to_dict() + async def calculate_simple_metrics(self, input_data: dict[str, Any]) -> dict[Hashable, Any]: + """ + Calculate simple metrics for a model. Metrics available are: + - rmse + - mse + - mae + - r2 + - accuracy + - precision + - recall + - f1 + Args: + input_data (dict[str, Any]): Input data containing: + - metadata (dict): Workflow execution metadata + - model_id (str): ID of the MLFlow model + - target_data (pd.DataFrame): Target data for calculating metrics, containing target and prediction columns + - metrics (list[str]): List of metrics to calculate + Returns: + dict[Hashable, Any]: Dictionary containing the calculated metrics + """ + + metadata = input_data['metadata'] + model_id = input_data['model_id'] + target_data = DataFrame(input_data['target_data']) + metrics = input_data['metrics'] + interval_minutes = input_data['interval_minutes'] + + data_size = len(target_data) + + output_data = [] + + diff = target_data['target'] - target_data['prediction'] + diff_squared = diff ** 2 + + self.info(f'Calculating simple metrics for model {model_id}: {metrics}', metadata) + + for metric in metrics: + if metric == 'rmse': + output_data.append({ + 'metric': 'rmse', + 'value': np.sqrt(np.mean(diff_squared)) + }) + elif metric == 'mse': + output_data.append({ + 'metric': 'mse', + 'value': np.mean(diff_squared) + }) + elif metric == 'mae': + output_data.append({ + 'metric': 'mae', + 'value': np.mean(np.abs(diff)) + }) + elif metric == 'r2': + y_true = target_data['target'] + y_mean = np.mean(y_true) + + ss_res = np.sum(diff_squared) + ss_tot = np.sum((y_true - y_mean) ** 2) + + # Evita divisão por zero + if ss_tot == 0: + r2_score = 0.0 + else: + r2_score = 1 - (ss_res / ss_tot) + + output_data.append({ + 'metric': 'r2', + 'value': r2_score + }) + + data = DataFrame(output_data) + data['model_id'] = model_id + data['timestamp'] = target_data['timestamp'].max() + data['data_size'] = data_size + data['interval_minutes'] = interval_minutes + + self.debug(f'Simple metrics dataframe: Size {data.shape} \n{data.head(5).to_string()}', metadata) + + return data.to_dict() + + diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 3d65de4..6bfd779 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -51,6 +51,8 @@ with workflow.unsafe.imports_passed_through(): from laborious.workflows.minimal_retrain import MinimalRetrain from laborious.workflows.predictions_batch import PredictionsBatch from laborious.workflows.drift import Drift + from laborious.workflows.simple_metrics import SientiaMetrics + from laborious.workflows.sub_workflows.format_and_export_prediction import ( FormatAndExportPrediction, ) @@ -175,6 +177,22 @@ async def main(): workflow_task_poller_behavior=PollerBehaviorAutoscaling(), activity_task_poller_behavior=PollerBehaviorAutoscaling(), ), + Worker( + temporal_client, + task_queue='simple_metrics-queue', + workflows=[SimpleMetrics], + activities=[ + activities.load_custom_query, + activities.calculate_simple_metrics, + activities.export_data_to_postgres, + ], + max_concurrent_workflow_tasks=50, + max_concurrent_activities=50, + max_concurrent_local_activities=50, + max_cached_workflows=2, + workflow_task_poller_behavior=PollerBehaviorAutoscaling(), + activity_task_poller_behavior=PollerBehaviorAutoscaling(), + ), Worker( temporal_client, task_queue='predictions_batch-queue', diff --git a/laborious/workflows/drift.py b/laborious/workflows/drift.py index d7c213c..7afaef7 100644 --- a/laborious/workflows/drift.py +++ b/laborious/workflows/drift.py @@ -80,7 +80,8 @@ class Drift: 'model_name': input_data['model_name'], 'model_id': input_data['model_id'], 'target_name': target_name, - 'drift_metrics': input_data['drift_metrics'], + 'drift_metrics': input_data.get('drift_metrics', + ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']), 'chunk_period': input_data.get('chunk_period', 'min'), }, retry_policy=retry_policy, diff --git a/laborious/workflows/simple_metrics.py b/laborious/workflows/simple_metrics.py new file mode 100644 index 0000000..f5f02d2 --- /dev/null +++ b/laborious/workflows/simple_metrics.py @@ -0,0 +1,95 @@ +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from datetime import timedelta + from typing import Any + + from sientia_do.temporal.policies import retry_policy + + from laborious.activities.activities import Activities + from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ + + +@workflow.defn(name='simple_metrics') +class SimpleMetrics: + @workflow.run + async def run(self, input_data: dict[str, Any]): + """ + Execute the simple metrics workflow. + """ + metadata = { + 'metadata': { + 'schedule_name': input_data['schedule_name'], + 'model_name': input_data['model_name'], + 'model_id': input_data['model_id'], + 'workflow_name': 'simple_metrics', + } + } + + model_id = input_data['model_id'] + interval_minutes = input_data['interval_minutes'] + + model_config = input_data['model_config'] + target_name = model_config['target'] + + query = f""" + select p."timestamp", p.prediction, ld.value as "target" + from {input_data['schema']}.{input_data['predictions_table_name']} p + inner join {input_data['schema']}.{input_data['data_table_name']} ld + on p."timestamp" = ld."timestamp" + where + p.model_id = {model_id} and + p.prediction is not null and + ld.variable = '{target_name}' and + ld.value is not null and + p."timestamp" >= NOW() - INTERVAL '{interval_minutes} minutes' + order by + p."timestamp" desc; + """ + + target_data = await workflow.execute_local_activity_method( + Activities.load_custom_query, + { + **metadata, + 'query': query, + 'datetime_columns': ['timestamp'], + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=300), + ) + + if not target_data: + return + + simple_metrics = await workflow.execute_local_activity_method( + Activities.calculate_simple_metrics, + { + **metadata, + 'model_id': model_id, + 'target_data': target_data, + 'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']), + 'interval_minutes': interval_minutes, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=300), + ) + + + if not simple_metrics: + return + + await workflow.execute_activity_method( + Activities.export_data_to_postgres, + { + **metadata, + 'data': simple_metrics, + 'schema': input_data['schema'], + 'table_name': input_data['target_table_name'], + 'timestamp_conversion': { + 'column': 'timestamp', + 'format': DATETIME_FORMAT_WITH_TZ, + }, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=300), + ) \ No newline at end of file diff --git a/tests/laborious/activities/test_model_metrics.py b/tests/laborious/activities/test_model_metrics.py index 8b55df8..87c3a68 100644 --- a/tests/laborious/activities/test_model_metrics.py +++ b/tests/laborious/activities/test_model_metrics.py @@ -639,3 +639,255 @@ async def test_get_drift_metrics_univariate_error( else: raise AssertionError('Expected Exception') + +@mark.asyncio +async def test_calculate_simple_metrics_success_all_metrics( + model_metrics_activity +): + # Arrange + target_data = DataFrame({ + 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'], + 'target': [1.0, 2.0, 3.0], + 'prediction': [1.1, 2.1, 2.9], + }) + + input_data = { + **metadata, + 'model_id': 'test_model_id', + 'target_data': target_data.to_dict(), + 'metrics': ['rmse', 'mse', 'mae', 'r2'], + 'interval_minutes': 5, + } + + # Act + result = DataFrame(await model_metrics_activity.calculate_simple_metrics(input_data)) + + # Assert + assert len(result['metric']) == 4 + assert 'rmse' in result['metric'].values + assert 'mse' in result['metric'].values + assert 'mae' in result['metric'].values + assert 'r2' in result['metric'].values + assert all(model_id == 'test_model_id' for model_id in result['model_id'].values) + assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values) + assert all(data_size == 3 for data_size in result['data_size'].values) + assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values) + model_metrics_activity.info.assert_called_once_with( + 'Calculating simple metrics for model test_model_id: [\'rmse\', \'mse\', \'mae\', \'r2\']', + metadata['metadata'] + ) + model_metrics_activity.debug.assert_called_once() + + +@mark.asyncio +async def test_calculate_simple_metrics_success_rmse_only( + model_metrics_activity +): + # Arrange + target_data = DataFrame({ + 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], + 'target': [1.0, 2.0], + 'prediction': [1.1, 2.1], + }) + + input_data = { + **metadata, + 'model_id': 'test_model_id', + 'target_data': target_data.to_dict(), + 'metrics': ['rmse'], + 'interval_minutes': 5, + } + + # Act + result = DataFrame(await model_metrics_activity.calculate_simple_metrics(input_data)) + + # Assert + assert len(result['metric']) == 1 + assert result['metric'].values[0] == 'rmse' + assert result['model_id'].values[0] == 'test_model_id' + assert result['timestamp'].values[0] == '2023-05-26 11:12:28' + assert result['data_size'].values[0] == 2 + assert result['interval_minutes'].values[0] == 5 + model_metrics_activity.info.assert_called_once_with( + 'Calculating simple metrics for model test_model_id: [\'rmse\']', + metadata['metadata'] + ) + + +@mark.asyncio +async def test_calculate_simple_metrics_success_mse_only( + model_metrics_activity +): + # Arrange + target_data = DataFrame({ + 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], + 'target': [1.0, 2.0], + 'prediction': [1.1, 2.1], + }) + + input_data = { + **metadata, + 'model_id': 'test_model_id', + 'target_data': target_data.to_dict(), + 'metrics': ['mse'], + 'interval_minutes': 5, + } + + # Act + result = DataFrame(await model_metrics_activity.calculate_simple_metrics(input_data)) + + # Assert + assert len(result['metric']) == 1 + assert result['metric'].values[0] == 'mse' + assert result['model_id'].values[0] == 'test_model_id' + assert result['timestamp'].values[0] == '2023-05-26 11:12:28' + assert result['data_size'].values[0] == 2 + assert result['interval_minutes'].values[0] == 5 + model_metrics_activity.info.assert_called_once_with( + 'Calculating simple metrics for model test_model_id: [\'mse\']', + metadata['metadata'] + ) + + +@mark.asyncio +async def test_calculate_simple_metrics_success_mae_only( + model_metrics_activity +): + # Arrange + target_data = DataFrame({ + 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], + 'target': [1.0, 2.0], + 'prediction': [1.1, 2.1], + }) + + input_data = { + **metadata, + 'model_id': 'test_model_id', + 'target_data': target_data.to_dict(), + 'metrics': ['mae'], + 'interval_minutes': 5, + } + + # Act + result = DataFrame(await model_metrics_activity.calculate_simple_metrics(input_data)) + + # Assert + assert len(result['metric']) == 1 + assert result['metric'].values[0] == 'mae' + assert result['model_id'].values[0] == 'test_model_id' + assert result['timestamp'].values[0] == '2023-05-26 11:12:28' + assert result['data_size'].values[0] == 2 + assert result['interval_minutes'].values[0] == 5 + model_metrics_activity.info.assert_called_once_with( + 'Calculating simple metrics for model test_model_id: [\'mae\']', + metadata['metadata'] + ) + + +@mark.asyncio +async def test_calculate_simple_metrics_success_r2_only( + model_metrics_activity +): + # Arrange + target_data = DataFrame({ + 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], + 'target': [1.0, 2.0], + 'prediction': [1.1, 2.1], + }) + + input_data = { + **metadata, + 'model_id': 'test_model_id', + 'target_data': target_data.to_dict(), + 'metrics': ['r2'], + 'interval_minutes': 5, + } + + # Act + result = DataFrame(await model_metrics_activity.calculate_simple_metrics(input_data)) + + # Assert + assert len(result['metric']) == 1 + assert result['metric'].values[0] == 'r2' + assert result['model_id'].values[0] == 'test_model_id' + assert result['timestamp'].values[0] == '2023-05-26 11:12:28' + assert result['data_size'].values[0] == 2 + assert result['interval_minutes'].values[0] == 5 + model_metrics_activity.info.assert_called_once_with( + 'Calculating simple metrics for model test_model_id: [\'r2\']', + metadata['metadata'] + ) + + +@mark.asyncio +async def test_calculate_simple_metrics_r2_zero_ss_tot( + model_metrics_activity +): + # Arrange + # All target values are the same, so ss_tot will be 0 + target_data = DataFrame({ + 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], + 'target': [1.0, 1.0], + 'prediction': [1.1, 1.1], + }) + + input_data = { + **metadata, + 'model_id': 'test_model_id', + 'target_data': target_data.to_dict(), + 'metrics': ['r2'], + 'interval_minutes': 5, + } + + # Act + result = DataFrame(await model_metrics_activity.calculate_simple_metrics(input_data)) + + # Assert + assert len(result['metric']) == 1 + assert result['metric'].values[0] == 'r2' + assert result['value'].values[0] == 0.0 # Should return 0.0 when ss_tot == 0 + assert result['model_id'].values[0] == 'test_model_id' + assert result['timestamp'].values[0] == '2023-05-26 11:12:28' + assert result['data_size'].values[0] == 2 + assert result['interval_minutes'].values[0] == 5 + model_metrics_activity.info.assert_called_once_with( + 'Calculating simple metrics for model test_model_id: [\'r2\']', + metadata['metadata'] + ) + + +@mark.asyncio +async def test_calculate_simple_metrics_success_multiple_metrics_subset( + model_metrics_activity +): + # Arrange + target_data = DataFrame({ + 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'], + 'target': [1.0, 2.0, 3.0], + 'prediction': [1.1, 2.1, 2.9], + }) + + input_data = { + **metadata, + 'model_id': 'test_model_id', + 'target_data': target_data.to_dict(), + 'metrics': ['rmse', 'mae'], + 'interval_minutes': 5, + } + + # Act + result = DataFrame(await model_metrics_activity.calculate_simple_metrics(input_data)) + + # Assert + assert len(result['metric']) == 2 + assert 'rmse' in result['metric'].values + assert 'mae' in result['metric'].values + assert all(model_id == 'test_model_id' for model_id in result['model_id'].values) + assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values) + assert all(data_size == 3 for data_size in result['data_size'].values) + assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values) + model_metrics_activity.info.assert_called_once_with( + 'Calculating simple metrics for model test_model_id: [\'rmse\', \'mae\']', + metadata['metadata'] + ) + diff --git a/tests/laborious/workflows/test_drift.py b/tests/laborious/workflows/test_drift.py index 82c9eb3..2f3f1dc 100644 --- a/tests/laborious/workflows/test_drift.py +++ b/tests/laborious/workflows/test_drift.py @@ -39,6 +39,8 @@ async def test_run(workflow_mock: AsyncMock, drift: Drift): 'chunk_period': 'hour', } + target_name = input_data['model_config']['target'] + target_data = {'data': 'test_target_data'} reference_data = {'data': 'test_reference_data'} drift_data = {'drift': 'test_drift_data'} @@ -96,7 +98,7 @@ async def test_run(workflow_mock: AsyncMock, drift: Drift): 'reference_data': reference_data, 'model_name': input_data['model_name'], 'model_id': input_data['model_id'], - 'target_name': input_data['target_name'], + 'target_name': target_name, 'drift_metrics': input_data['drift_metrics'], 'chunk_period': input_data['chunk_period'], }, @@ -131,7 +133,7 @@ async def test_run_empty_target_data(workflow_mock: AsyncMock, drift: Drift): 'source_table_name': 'test_source_table', 'target_table_name': 'test_target_table', 'interval': 60, - 'target_name': 'test_target', + 'model_config': {'target': 'test_target'}, 'drift_metrics': ['psi', 'ks'], } @@ -163,10 +165,12 @@ async def test_run_empty_drift_data(workflow_mock: AsyncMock, drift: Drift): 'source_table_name': 'test_source_table', 'target_table_name': 'test_target_table', 'interval': 60, - 'target_name': 'test_target', + 'model_config': {'target': 'test_target'}, 'drift_metrics': ['psi', 'ks'], } + target_name = input_data['model_config']['target'] + target_data = {'data': 'test_target_data'} reference_data = {'data': 'test_reference_data'} drift_data = None @@ -188,7 +192,7 @@ async def test_run_empty_drift_data(workflow_mock: AsyncMock, drift: Drift): 'reference_data': reference_data, 'model_name': input_data['model_name'], 'model_id': input_data['model_id'], - 'target_name': input_data['target_name'], + 'target_name': target_name, 'drift_metrics': input_data['drift_metrics'], 'chunk_period': input_data.get('chunk_period', 'min'), }, @@ -211,11 +215,13 @@ async def test_run_default_chunk_period(workflow_mock: AsyncMock, drift: Drift): 'source_table_name': 'test_source_table', 'target_table_name': 'test_target_table', 'interval': 60, - 'target_name': 'test_target', + 'model_config': {'target': 'test_target'}, 'drift_metrics': ['psi', 'ks'], # chunk_period not provided, should default to 'min' } + target_name = input_data['model_config']['target'] + target_data = {'data': 'test_target_data'} reference_data = {'data': 'test_reference_data'} drift_data = {'drift': 'test_drift_data'} @@ -237,7 +243,7 @@ async def test_run_default_chunk_period(workflow_mock: AsyncMock, drift: Drift): 'reference_data': reference_data, 'model_name': input_data['model_name'], 'model_id': input_data['model_id'], - 'target_name': input_data['target_name'], + 'target_name': target_name, 'drift_metrics': input_data['drift_metrics'], 'chunk_period': 'min', # Default value }, diff --git a/tests/laborious/workflows/test_simple_metrics.py b/tests/laborious/workflows/test_simple_metrics.py new file mode 100644 index 0000000..b185b1a --- /dev/null +++ b/tests/laborious/workflows/test_simple_metrics.py @@ -0,0 +1,228 @@ +from unittest.mock import ANY, AsyncMock, call, patch + +from pytest import fixture, mark + +from laborious.activities.activities import Activities +from laborious.workflows.simple_metrics import SimpleMetrics +from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ + + +@fixture +def simple_metrics() -> SimpleMetrics: + return SimpleMetrics() + + +metadata = { + 'metadata': { + 'model_id': 'test_model_id', + 'model_name': 'test_model', + 'workflow_name': 'simple_metrics', + 'schedule_name': 'test_schedule', + }, +} + + +@mark.asyncio +@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) +async def test_run(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics): + # Arrange + input_data = { + 'schedule_name': 'test_schedule', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'interval_minutes': 60, + 'model_config': {'target': 'test_target'}, + 'schema': 'test_schema', + 'predictions_table_name': 'test_predictions_table', + 'data_table_name': 'test_data_table', + 'target_table_name': 'test_target_table', + 'metrics': ['rmse', 'mse', 'mae', 'r2'], + } + + target_data = {'data': 'test_target_data'} + simple_metrics_data = {'metrics': 'test_simple_metrics_data'} + + workflow_mock.execute_local_activity_method.side_effect = [ + target_data, simple_metrics_data + ] + + workflow_mock.execute_activity_method = AsyncMock() + + # Act + await simple_metrics.run(input_data) + + # Assert - Check load_custom_query call + expected_query = f""" + select p."timestamp", p.prediction, ld.value as "target" + from {input_data['schema']}.{input_data['predictions_table_name']} p + inner join {input_data['schema']}.{input_data['data_table_name']} ld + on p."timestamp" = ld."timestamp" + where + p.model_id = {input_data['model_id']} and + p.prediction is not null and + ld.variable = '{input_data['model_config']['target']}' and + ld.value is not null and + p."timestamp" >= NOW() - INTERVAL '{input_data['interval_minutes']} minutes' + order by + p."timestamp" desc; + """ + + workflow_mock.execute_local_activity_method.assert_has_calls( + [ + call( + Activities.load_custom_query, + { + **metadata, + 'query': expected_query, + 'datetime_columns': ['timestamp'], + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ), + call( + Activities.calculate_simple_metrics, + { + **metadata, + 'model_id': input_data['model_id'], + 'target_data': target_data, + 'metrics': input_data['metrics'], + 'interval_minutes': input_data['interval_minutes'], + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ), + ] + ) + + # Assert - Check export_data_to_postgres call + workflow_mock.execute_activity_method.assert_called_once_with( + Activities.export_data_to_postgres, + { + **metadata, + 'data': simple_metrics_data, + 'schema': input_data['schema'], + 'table_name': input_data['target_table_name'], + 'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ}, + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ) + + +@mark.asyncio +@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) +async def test_run_empty_target_data(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics): + # Arrange + input_data = { + 'schedule_name': 'test_schedule', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'interval_minutes': 60, + 'model_config': {'target': 'test_target'}, + 'schema': 'test_schema', + 'predictions_table_name': 'test_predictions_table', + 'data_table_name': 'test_data_table', + 'target_table_name': 'test_target_table', + 'metrics': ['rmse', 'mse'], + } + + target_data = None + + workflow_mock.execute_local_activity_method.return_value = target_data + workflow_mock.execute_activity_method = AsyncMock() + + # Act + await simple_metrics.run(input_data) + + # Assert - Should not call calculate_simple_metrics or export + assert workflow_mock.execute_local_activity_method.call_count == 1 + workflow_mock.execute_activity_method.assert_not_called() + + +@mark.asyncio +@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) +async def test_run_empty_simple_metrics(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics): + # Arrange + input_data = { + 'schedule_name': 'test_schedule', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'interval_minutes': 60, + 'model_config': {'target': 'test_target'}, + 'schema': 'test_schema', + 'predictions_table_name': 'test_predictions_table', + 'data_table_name': 'test_data_table', + 'target_table_name': 'test_target_table', + 'metrics': ['rmse', 'mse'], + } + + target_data = {'data': 'test_target_data'} + simple_metrics_data = None + + workflow_mock.execute_local_activity_method.side_effect = [ + target_data, simple_metrics_data + ] + + workflow_mock.execute_activity_method = AsyncMock() + + # Act + await simple_metrics.run(input_data) + + # Assert - Should call calculate_simple_metrics but not export + assert workflow_mock.execute_local_activity_method.call_count == 2 + workflow_mock.execute_activity_method.assert_not_called() + + +@mark.asyncio +@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) +async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics): + # Arrange + input_data = { + 'schedule_name': 'test_schedule', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'interval_minutes': 60, + 'model_config': {'target': 'test_target'}, + 'schema': 'test_schema', + 'predictions_table_name': 'test_predictions_table', + 'data_table_name': 'test_data_table', + 'target_table_name': 'test_target_table', + # metrics not provided, should default to ['rmse', 'mse', 'mae', 'r2'] + } + + target_data = {'data': 'test_target_data'} + simple_metrics_data = {'metrics': 'test_simple_metrics_data'} + + workflow_mock.execute_local_activity_method.side_effect = [ + target_data, simple_metrics_data + ] + + workflow_mock.execute_activity_method = AsyncMock() + + # Act + await simple_metrics.run(input_data) + + # Assert - Check calculate_simple_metrics call with default metrics + workflow_mock.execute_local_activity_method.assert_has_calls( + [ + call( + Activities.load_custom_query, + ANY, + retry_policy=ANY, + start_to_close_timeout=ANY, + ), + call( + Activities.calculate_simple_metrics, + { + **metadata, + 'model_id': input_data['model_id'], + 'target_data': target_data, + 'metrics': ['rmse', 'mse', 'mae', 'r2'], # Default value + 'interval_minutes': input_data['interval_minutes'], + }, + retry_policy=ANY, + start_to_close_timeout=ANY, + ), + ] + ) + diff --git a/values.yaml b/values.yaml index f8ad89e..5034e12 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: "1.1.0" + tag: "1.1.1" 0# 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: @@ -151,7 +151,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" - name: GITHUB_BRANCH - value: "feature/SIENTIAPDE-1325-adicionar-metricas-especificas-de-operacoes-externas" + value: "feature/SIENTIAPDE-1273" - name: PYTHON_APP value: "laborious.worker.worker" @@ -234,7 +234,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-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.5.0 +# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.0 # kubectl create secret generic git-ssh-key-sientia-laborious-worker \ # --namespace sientia \