Merge pull request #21 from Aignosi/SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados

SIENTIAPDE-1312/SIENTIAPDE-1318: Implement 'fill_missing_tags' parameter and update image tag

scouter/activities/redis.py

@@ -185,6 +185,7 @@ class Redis(RedisBase):
         data = DataFrame(input_data['data'])
         model_tags = input_data['model_tags']
         retention_time = input_data['retention_time']
+        fill_missing_tags = input_data['fill_missing_tags']
 
         key = f'held_data_{input_data["workflow_name"]}_{input_data["schedule_name"]}'
 
@@ -226,6 +227,13 @@ class Redis(RedisBase):
                 data_hold[row['name']] = value
                 to_register_metrics.append((row['name'], value))
 
+            if fill_missing_tags:
+                self.debug('Filling missing tags in data package', metadata=metadata)
+                missing_tags = [tag for tag in tags if tag not in list(data_hold.keys())]
+
+                for tag in missing_tags:
+                    data_hold[tag] = None
+
             data_hold['timestamp'] = (
                 data['timestamp'].max() if not data.empty else data_hold['timestamp']
             )

scouter/workflow/sub_workflows/core_scouter.py

@@ -93,6 +93,7 @@ class CoreScouter:
                 'model_id': input_data['model_id'],
                 'model_tags': input_data['model_tags'],
                 'retention_time': input_data['retention_time'],
+                'fill_missing_tags': input_data['fill_missing_tags'],
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60),

tests/activities/test_redis.py

@@ -216,6 +216,7 @@ async def test_group_and_hold_data_new_key(redis_activity):
             }
         ).to_dict('records'),
         'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'},
+        'fill_missing_tags': False,
     }
 
     # Mock get to return None for new key
@@ -243,7 +244,7 @@ async def test_group_and_hold_data_new_key(redis_activity):
 
 
 @pytest.mark.asyncio
-async def test_group_and_hold_data_update_existing(redis_activity):
+async def test_group_and_hold_data_update_existing_fill_missing(redis_activity):
     """Test updating existing data with group_and_hold_data"""
     # Setup initial data in Redis
     existing_data = {'sensor1': 20.0, 'sensor2': 28.0, 'timestamp': '2023-01-01 11:00:00'}
@@ -262,7 +263,13 @@ async def test_group_and_hold_data_update_existing(redis_activity):
                 'timestamp': ['2023-01-01 12:00:00'] * 2,
             }
         ).to_dict('records'),
-        'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2', 'sensor3': 'sensor3'},
+        'model_tags': {
+            'sensor1': 'sensor1',
+            'sensor2': 'sensor2',
+            'sensor3': 'sensor3',
+            'sensor4': 'sensor4',
+        },
+        'fill_missing_tags': True,
     }
 
     # Mock get to return existing data
@@ -274,10 +281,15 @@ async def test_group_and_hold_data_update_existing(redis_activity):
 
     # Verify the result
     expected_result = {
-        'timestamp': {0: '2023-01-01 12:00:00', 1: '2023-01-01 12:00:00', 2: '2023-01-01 12:00:00'},
+        'timestamp': {
-        'variable': {0: 'sensor1', 1: 'sensor2', 2: 'sensor3'},
+            0: '2023-01-01 12:00:00',
-        'value': {0: 25.5, 1: 28.0, 2: 42.0},
+            1: '2023-01-01 12:00:00',
-        'model_id': {0: 1, 1: 1, 2: 1},
+            2: '2023-01-01 12:00:00',
+            3: '2023-01-01 12:00:00',
+        },
+        'variable': {0: 'sensor1', 1: 'sensor2', 2: 'sensor3', 3: 'sensor4'},
+        'value': {0: 25.5, 1: 28.0, 2: 42.0, 3: None},
+        'model_id': {0: 1, 1: 1, 2: 1, 3: 1},
     }
     assert result == expected_result
 
@@ -289,6 +301,7 @@ async def test_group_and_hold_data_update_existing(redis_activity):
         'sensor1': 25.5,
         'sensor2': 28.0,
         'sensor3': 42.0,
+        'sensor4': None,
         'timestamp': '2023-01-01 12:00:00',
     }
     assert kwargs['ttl'] == 3600
@@ -312,6 +325,7 @@ async def test_group_and_hold_data_with_none_values(redis_activity):
             }
         ).to_dict('records'),
         'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'},
+        'fill_missing_tags': False,
     }
 
     # Mock get to return None for new key
@@ -337,6 +351,7 @@ async def test_group_and_hold_data_empty_dataframe(redis_activity):
         'retention_time': 3600,
         'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'),
         'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'},
+        'fill_missing_tags': False,
     }
 
     redis_activity.get = MagicMock(return_value=None)
@@ -358,6 +373,7 @@ async def test_group_and_hold_data_error_get(redis_activity):
         'model_id': 1,
         'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'),
         'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'},
+        'fill_missing_tags': False,
     }
 
     redis_activity.get = MagicMock(side_effect=Exception('test'))
@@ -393,6 +409,7 @@ async def test_group_and_hold_data_error_set(redis_activity):
         'model_id': 1,
         'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'),
         'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'},
+        'fill_missing_tags': False,
     }
 
     existing_data = {'sensor1': 20.0, 'sensor2': 28.0, 'timestamp': '2023-01-01 11:00:00'}

tests/workflow/sub_workflows/test_core_scouter.py

@@ -42,6 +42,7 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter):
             'retention_time': 3600,
             'model_tags': {},
             'debug_data_package': True,
+            'fill_missing_tags': False,
         }
     )
 
@@ -91,6 +92,7 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter):
                     'model_id': 'test_model_id',
                     'retention_time': 3600,
                     'model_tags': {},
+                    'fill_missing_tags': False,
                 },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
@@ -161,6 +163,7 @@ async def test_core_scouter_workflow_with_empty_data(mock_workflow, core_scouter
             'table_name': 'test_table',
             'retention_time': 3600,
             'model_tags': {},
+            'fill_missing_tags': False,
         }
     )
 
@@ -210,6 +213,7 @@ async def test_core_scouter_workflow_with_empty_data(mock_workflow, core_scouter
                     'model_id': 'test_model_id',
                     'retention_time': 3600,
                     'model_tags': {},
+                    'fill_missing_tags': False,
                 },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,

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.4.5"
+  tag: "0.4.9"
 
 # 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:
@@ -150,7 +150,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git"
   - name: GITHUB_BRANCH
-    value: "main"
+    value: "SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados"
   - name: PYTHON_APP
     value: "scouter.worker.worker"