Merge pull request #44 from Aignosi/fix/SIENTIAPDE-1993

fix: preserve real timestamps in get_tag_values when max_count > 1
2026-09-14 09:13:23 -03:00
parent 554a26bf2c 1d764439cb
commit faca362fb8
7 changed files with 179 additions and 13 deletions
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -48,7 +48,7 @@ Parent/child Temporal workflow pattern, two independent worker task queues shari
 ```
 Scouter (scouter/workflow/scouter.py)              PIWebAPIScouter (scouter/workflow/pi_web_api_scouter.py)
  reads raw_{schedule_name} from MongoDB              queries PI Web API directly (get_tag_values)
-  tracks last-processed timestamp in Redis             normalizes all timestamps to the batch max
+  tracks last-processed timestamp in Redis             preserves each record's real timestamp
        │                                                     │
        └───────────────► CoreScouter (scouter/workflow/sub_workflows/core_scouter.py) ◄───────┘
                              1. data_quality_gate   (NULL_VALUES_FILTER / OUT_OF_BOUNDS_FILTER, DISCARD policy)
--- a/e2e/scenario_inputs/pi_web_api_scouter_distinct_tag_timestamps_multi_point.json
+++ b/e2e/scenario_inputs/pi_web_api_scouter_distinct_tag_timestamps_multi_point.json
@@ -0,0 +1,36 @@
+{
+  "workflow_input": {
+    "model_id": "13",
+    "model_name": "PI Distinct Tag Timestamps (max_count>1)",
+    "schedule_name": "pi-distinct-ts-n",
+    "trigger_laborious": false,
+    "filters": {},
+    "schema": "sientia_data",
+    "table_name": "laborious_data",
+    "retention_time": 3600,
+    "fill_missing_tags": false,
+    "model_tags": {
+      "tag_early": {"webid": "w1", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000},
+      "tag_late": {"webid": "w2", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000}
+    },
+    "pi_web_api_query": {
+      "endpoint": "/streamsets/recorded",
+      "period": "*-1d",
+      "max_count": 2,
+      "api_timeout": 30
+    }
+  },
+  "pi_web_api_server": {
+    "mode": "success",
+    "rows": [
+      {"name": "tag_early", "timestamp": "2024-06-01T12:00:00Z", "value": 10},
+      {"name": "tag_early", "timestamp": "2024-06-01T12:01:00Z", "value": 20},
+      {"name": "tag_late", "timestamp": "2024-06-01T15:00:00Z", "value": 100},
+      {"name": "tag_late", "timestamp": "2024-06-01T15:05:00Z", "value": 200}
+    ]
+  },
+  "expected_rows": {
+    "tag_early": {"value": 20.0, "timestamp": "2024-06-01T12:01:00Z"},
+    "tag_late": {"value": 200.0, "timestamp": "2024-06-01T15:05:00Z"}
+  }
+}
--- a/e2e/scenario_inputs/pi_web_api_scouter_distinct_tag_timestamps_single_point.json
+++ b/e2e/scenario_inputs/pi_web_api_scouter_distinct_tag_timestamps_single_point.json
@@ -0,0 +1,34 @@
+{
+  "workflow_input": {
+    "model_id": "14",
+    "model_name": "PI Distinct Tag Timestamps (max_count=1)",
+    "schedule_name": "pi-distinct-ts-1",
+    "trigger_laborious": false,
+    "filters": {},
+    "schema": "sientia_data",
+    "table_name": "laborious_data",
+    "retention_time": 3600,
+    "fill_missing_tags": false,
+    "model_tags": {
+      "tag_early": {"webid": "w1", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000},
+      "tag_late": {"webid": "w2", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000}
+    },
+    "pi_web_api_query": {
+      "endpoint": "/streamsets/recorded",
+      "period": "*-1d",
+      "max_count": 1,
+      "api_timeout": 30
+    }
+  },
+  "pi_web_api_server": {
+    "mode": "success",
+    "rows": [
+      {"name": "tag_early", "timestamp": "2024-06-01T12:00:00Z", "value": 10},
+      {"name": "tag_late", "timestamp": "2024-06-01T15:00:00Z", "value": 100}
+    ]
+  },
+  "expected_rows": {
+    "tag_early": {"value": 10.0, "timestamp": "2024-06-01T12:00:00Z"},
+    "tag_late": {"value": 100.0, "timestamp": "2024-06-01T15:00:00Z"}
+  }
+}
--- a/e2e/scenarios.md
+++ b/e2e/scenarios.md
@@ -61,6 +61,14 @@ Five tags with `avg` / `mdn` / `max` / `min` / `lts`; assert five distinct `vari

 `debug_data_package=True`; assert `data_package_pi_web_api_scouter_*` Redis key with `data` and `held_data`.

+### 2.1.4 Distinct tag timestamps preserved
+
+Regression test for SIENTIAPDE-1993, parametrized over `max_count=1` and `max_count>1`. Two
+tags (`tag_early` around 12:00, `tag_late` around 15:00) per case; assert each tag's stored
+`timestamp` in Postgres equals its own latest point after aggregation, not the other tag's
+(the batch-wide max across tags) — the invariant must hold whether a tag has one point or
+several historical points.
+
 ### 2.2.1 Empty response early exit

 Server `empty` mode; zero Postgres rows for `model_id`, one request recorded.
--- a/e2e/test_pi_web_api_scouter_main_workflow.py
+++ b/e2e/test_pi_web_api_scouter_main_workflow.py
@@ -3,6 +3,7 @@ End-to-end tests for the PIWebAPIScouter main workflow.
 """

 import json
+from datetime import datetime

 import pytest
 from redis import Redis
@@ -103,6 +104,52 @@ async def test_scenario_2_1_3_debug_data_package(
    assert 'data' in payload and 'held_data' in payload


+@pytest.mark.e2e
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'scenario_name, workflow_id_prefix',
+    [
+        ('pi_web_api_scouter_distinct_tag_timestamps_single_point', 'pi-distinct-ts-1'),
+        ('pi_web_api_scouter_distinct_tag_timestamps_multi_point', 'pi-distinct-ts-n'),
+    ],
+    ids=['max_count_1', 'max_count_gt_1'],
+)
+async def test_scenario_2_1_4_distinct_tag_timestamps_preserved(
+    temporal_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    pi_web_api_server: PIWebAPITestServer,
+    postgres_engine,
+    scenario_name: str,
+    workflow_id_prefix: str,
+):
+    """
+    Regression test for SIENTIAPDE-1993: whether a tag has a single point
+    (max_count=1) or several historical points (max_count>1), each tag's
+    stored row must keep the timestamp of its own latest point after
+    aggregation, not another tag's timestamp from the same batch.
+    """
+    scenario = load_scenario_input(scenario_name)
+    apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
+    workflow_input = scenario['workflow_input']
+
+    await start_and_await_workflow(
+        temporal_env.client,
+        PIWebAPIScouter.run,
+        workflow_input,
+        make_workflow_id(workflow_id_prefix),
+    )
+
+    rows = {
+        row['variable']: row
+        for row in fetch_laborious_rows(postgres_engine, workflow_input['model_id'])
+    }
+    for tag, expected in scenario['expected_rows'].items():
+        assert tag in rows
+        assert rows[tag]['value'] == pytest.approx(expected['value'])
+        expected_timestamp = datetime.fromisoformat(expected['timestamp'].replace('Z', '+00:00'))
+        assert rows[tag]['timestamp'] == expected_timestamp
+
+
@pytest.mark.e2e
@pytest.mark.asyncio
@pytest.mark.xfail(
--- a/scouter/activities/api.py
+++ b/scouter/activities/api.py
@@ -86,10 +86,8 @@ class API(SientiaMonitoring):
        for a set of configured tags. It returns the data as a list of dictionaries
        suitable for further processing in the workflow.

-        The timestamps are normalized to ensure consistency across all records in the
-        response. After converting timestamps to string format, all timestamps are
-        set to the maximum timestamp value (lexicographically) found in the dataset.
-        This ensures all records in a single batch share the same timestamp value.
+        Timestamps are converted to string format and each record keeps its own real
+        timestamp as returned by the PI Web API.

        Args:
            input_data (dict[str, Any]): Activity input parameters.
@@ -103,7 +101,7 @@ class API(SientiaMonitoring):

        Returns:
            list[dict]: List of data records, each containing:
-                - timestamp: Normalized timestamp string (all records share the same value)
+                - timestamp: Each record's real timestamp string
                - name: Tag name
                - value: Numeric value
                - tag: WebId
@@ -148,10 +146,6 @@ class API(SientiaMonitoring):

        self.debug(f'Latest values: {latest_values.to_string()}', metadata=metadata)

-        # Normalize the package timestamp
-        valid_timestamp_values = latest_values['timestamp'].dropna()
-        latest_values['timestamp'] = valid_timestamp_values.max()
-
        self.info(f'Gathered {len(latest_values)} tag values', metadata=metadata)

        return latest_values.to_dict(orient='records')
--- a/tests/activities/test_api.py
+++ b/tests/activities/test_api.py
@@ -87,7 +87,7 @@ def test_close(mock_sientia_monitoring, api_activity):


 def test_get_tag_values_success(api_activity):
-    """Test get_tag_values with successful data retrieval."""
+    """Test get_tag_values with successful data retrieval preserves each record's real timestamp."""
    # Setup test data
    test_data = {
        **metadata,
@@ -155,8 +155,55 @@ def test_get_tag_values_success(api_activity):
    assert result[0]['value'] == pytest.approx(10.5)
    assert result[1]['name'] == 'tag2'
    assert result[2]['name'] == 'tag3'
-    assert result[0]['timestamp'] == '2023-01-01 12:02:00+0000'
-    assert result[1]['timestamp'] == '2023-01-01 12:02:00+0000'
+    # Each record keeps its own real timestamp, no cross-tag normalization
+    assert result[0]['timestamp'] == '2023-01-01 12:00:00+0000'
+    assert result[1]['timestamp'] == '2023-01-01 12:01:00+0000'
+    assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000'
+
+
+def test_get_tag_values_with_max_count_greater_than_one_preserves_timestamps(api_activity):
+    """Regression test for SIENTIAPDE-1993: with max_count > 1, each historical point's
+    real timestamp must be preserved, otherwise distinct records collide on the
+    (model_id, timestamp, variable) unique constraint downstream and are silently
+    dropped."""
+    test_data = {
+        **metadata,
+        'endpoint': '/streamsets/recorded',
+        'web_ids': {
+            'tag1': {
+                'webid': 'webid1',
+                'aggr_function': 'avg',
+                'data_range': [0, 100],
+            },
+        },
+        'period': '*-1d',
+        'max_count': 3,
+        'api_timeout': 30,
+    }
+
+    # Multiple historical points for the SAME tag, with distinct timestamps
+    mock_df = pd.DataFrame(
+        {
+            'timestamp': [
+                '2023-01-01 12:00:00+0000',
+                '2023-01-01 12:01:00+0000',
+                '2023-01-01 12:02:00+0000',
+            ],
+            'name': ['tag1', 'tag1', 'tag1'],
+            'value': [10.5, 20.3, 30.7],
+            'tag': ['webid1', 'webid1', 'webid1'],
+        }
+    )
+
+    mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp'])
+
+    api_activity.pi_web_api_client.get_latest_values_df = Mock(return_value=mock_df)
+
+    result = api_activity.get_tag_values(test_data)
+
+    assert len(result) == 3
+    assert result[0]['timestamp'] == '2023-01-01 12:00:00+0000'
+    assert result[1]['timestamp'] == '2023-01-01 12:01:00+0000'
    assert result[2]['timestamp'] == '2023-01-01 12:02:00+0000'