diff --git a/CLAUDE.md b/CLAUDE.md index e24f23e..36e3d70 100644 --- 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) diff --git 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 new file mode 100644 index 0000000..e4c24de --- /dev/null +++ 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"} + } +} diff --git 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 new file mode 100644 index 0000000..f400051 --- /dev/null +++ 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"} + } +} diff --git a/e2e/scenarios.md b/e2e/scenarios.md index caac4c5..3c9f87e 100644 --- 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. diff --git a/e2e/test_pi_web_api_scouter_main_workflow.py b/e2e/test_pi_web_api_scouter_main_workflow.py index 5696310..3f7c447 100644 --- 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( diff --git a/scouter/activities/api.py b/scouter/activities/api.py index 161d3a6..32c087f 100644 --- 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') diff --git a/tests/activities/test_api.py b/tests/activities/test_api.py index 3c12209..878cc5d 100644 --- 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'