From 9e3a747b946c7ff9dc10d9c49ccc71d76aed0569 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Tue, 29 Sep 2026 17:06:54 +0100 Subject: [PATCH 1/4] Use ISO/UTC timestamps in the Waveform CSV and parquet files, to match the timestamps we're using in the EHR parquets. --- src/controller.py | 10 +++------ src/csv_writer.py | 5 ++--- src/emap_interchange/messages.py | 7 +++++-- src/exporter/daily_summary.py | 8 +++++++- src/exporter/parquet.py | 6 +++++- src/pseudon/pseudon.py | 7 ++++--- tests/helpers.py | 1 + tests/test_controller.py | 14 +++++++++---- tests/test_file_writer.py | 8 ++++---- tests/test_snakemake_integration.py | 32 ++++++++++++++++++++--------- 10 files changed, 63 insertions(+), 35 deletions(-) diff --git a/src/controller.py b/src/controller.py index 8f9fd1a..2bb4d85 100644 --- a/src/controller.py +++ b/src/controller.py @@ -4,7 +4,6 @@ """ from dataclasses import dataclass -from datetime import datetime, timezone import logging from typing import Literal, Any, Optional @@ -204,20 +203,17 @@ def outcome( ) return outcome("reject", reason="numeric_string_confusion", requeue=False) - observation_time = datetime.fromtimestamp( - observation_timestamp, tz=timezone.utc - ) lookup_success = True try: matched_mrn = self.emap_db.get_matched_mrn( - location_string, observation_time + location_string, observation_timestamp ) except ValueError: lookup_success = False logger.error( "Ambiguous or non existent match for location %s, obs time %s", location_string, - observation_time, + observation_timestamp, exc_info=True, ) matched_mrn = ("unmatched_mrn", "unmatched_nhs", "unmatched_csn", False) @@ -234,7 +230,7 @@ def outcome( source_variable_id=source_variable_id, source_channel_id=source_channel_id, sampling_rate=sampling_rate, - observation_timestamp=observation_timestamp, + observation_datetime=observation_timestamp, units=units, mapped_location_string=mapped_location_string, csn=csn, diff --git a/src/csv_writer.py b/src/csv_writer.py index b904ca7..cc085cb 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -42,7 +42,7 @@ def write_frame( string_values: Optional[list[str]] = None, source_variable_id: str, source_channel_id: Optional[str] = None, - observation_timestamp: float, + observation_datetime: datetime, units: str, sampling_rate: Optional[int] = None, mapped_location_string: str, @@ -59,7 +59,6 @@ def write_frame( raise ValueError( "Exactly ONE of string_values, numeric_values must be not None" ) - observation_datetime = datetime.fromtimestamp(observation_timestamp) filename = WAVEFORM_ORIGINAL_CSV / create_file_name( source_variable_id, source_channel_id, observation_datetime, csn, units @@ -90,7 +89,7 @@ def write_frame( source_channel_id if source_channel_id is not None else "", units, sampling_rate if sampling_rate is not None else "", - observation_timestamp, + observation_datetime.isoformat(), mapped_location_string, json.dumps(numeric_values) if numeric_values is not None else "", json.dumps(string_values) if string_values is not None else "", diff --git a/src/emap_interchange/messages.py b/src/emap_interchange/messages.py index b9380a7..4954671 100644 --- a/src/emap_interchange/messages.py +++ b/src/emap_interchange/messages.py @@ -3,6 +3,7 @@ # code as part of its build process, but for now it's just copied # and modified import json +from datetime import datetime, timezone class WaveformBaseMessage: @@ -29,9 +30,11 @@ def from_json(json_data) -> "WaveformBaseMessage": return message_cls(data) - def get_observation_time(self): + def get_observation_time(self) -> datetime: """Time of the observation.""" - return self.data["observationTime"] + # Message time is epoch seconds, so convert to a Python timestamp + timestamp_epoch_seconds = self.data["observationTime"] + return datetime.fromtimestamp(timestamp_epoch_seconds, tz=timezone.utc) def get_source_location_string(self): """Location string according to the original data source.""" diff --git a/src/exporter/daily_summary.py b/src/exporter/daily_summary.py index a067d78..d545789 100644 --- a/src/exporter/daily_summary.py +++ b/src/exporter/daily_summary.py @@ -1,5 +1,6 @@ import json import logging +from datetime import datetime from exporter.parquet import parquet_min_max_value @@ -41,6 +42,11 @@ def make_daily_hash_summary(daily_files, out_json_file): hash_summary = list(hash_summary_by_csn.values()) + def json_serialise(obj): + if isinstance(obj, datetime): + return obj.isoformat() + raise TypeError(f"Type {type(obj)} not serializable") + with open(out_json_file, "w") as fh: - json.dump(hash_summary, fh, indent=0) + json.dump(hash_summary, fh, indent=0, default=json_serialise) logger.info(f"Wrote {len(hash_summary)} entries to {out_json_file}") diff --git a/src/exporter/parquet.py b/src/exporter/parquet.py index 7362497..7dd353c 100644 --- a/src/exporter/parquet.py +++ b/src/exporter/parquet.py @@ -1,9 +1,13 @@ +from datetime import datetime from pathlib import Path +from typing import Tuple, Optional from pyarrow import parquet as pq -def parquet_min_max_value(parquet_path: Path, column_name): +def parquet_min_max_value( + parquet_path: Path, column_name +) -> Tuple[Optional[datetime], Optional[datetime]]: """By the magic of parquet files we can get the min/max timestamps without loading it all into memory or even reading every row.""" parquet_file = pq.ParquetFile(parquet_path) diff --git a/src/pseudon/pseudon.py b/src/pseudon/pseudon.py index 6cfa199..3e24620 100644 --- a/src/pseudon/pseudon.py +++ b/src/pseudon/pseudon.py @@ -99,14 +99,15 @@ def csv_to_parquets( "source_channel_id": str, "units": str, "sampling_rate": "Int32", - "timestamp": float, + "timestamp": str, "location": str, "numeric_values": str, "string_values": str, }, header=0, # the first line is always the header ) - + # go from ISO str to pandas timestamp + df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True) df["numeric_values"] = df["numeric_values"].apply(parse_numeric_values) df["string_values"] = df["string_values"].apply(parse_string_values) @@ -122,7 +123,7 @@ def csv_to_parquets( ("source_channel_id", pa.string()), ("units", pa.string()), ("sampling_rate", pa.int32()), - ("timestamp", pa.float64()), + ("timestamp", pa.timestamp("us", tz="UTC")), ("location", pa.string()), # As per requirements, compactness is important here. # decimal32 can have a maximum of 9 significant digits and should diff --git a/tests/helpers.py b/tests/helpers.py index 48159a2..75f3aa7 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -13,6 +13,7 @@ class TestFileDescription: __test__ = False date: str + # use POSIX timestamps because the arithmetic is a bit easier start_timestamp: float csn: str mrn: str diff --git a/tests/test_controller.py b/tests/test_controller.py index bcf6eb6..7c83fac 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1,6 +1,6 @@ import copy import json -from datetime import datetime +from datetime import datetime, timezone from typing import Literal from unittest.mock import Mock @@ -18,6 +18,12 @@ def __init__(self, missing_key, value_type): self.fake_data = FakeData._base_fake_data() self.value_type: Literal["numeric", "string", "both"] = value_type + # comes in as POSIX timestamp, out as Python datetime + fake_timestamp_io = [ + 1790614413.9698, + datetime(2026, 9, 28, 16, 53, 33, 969800, tzinfo=timezone.utc), + ] + @staticmethod def _base_fake_data() -> dict: return { @@ -27,7 +33,7 @@ def _base_fake_data() -> dict: "sourceObservationType": "waveform", "mappedVariableDescription": "P0.1 Occlusion Pressure", "mappedLocationString": "loc", - "observationTime": datetime.now().timestamp(), + "observationTime": FakeData.fake_timestamp_io[0], "sourceVariableId": "27", "unit": "uV", } @@ -57,7 +63,7 @@ def get_expected_write_frame_kwargs(self) -> dict: "string_values": None, # HF is always numeric "source_variable_id": fd["sourceVariableId"], "source_channel_id": fd["sourceChannelId"], - "observation_timestamp": fd["observationTime"], + "observation_datetime": FakeData.fake_timestamp_io[1], "units": fd["unit"], "sampling_rate": fd["samplingRate"], "mapped_location_string": fd["mappedLocationString"], @@ -116,7 +122,7 @@ def get_expected_write_frame_kwargs(self) -> dict: expected = { "source_variable_id": fd["sourceVariableId"], "source_channel_id": None, - "observation_timestamp": fd["observationTime"], + "observation_datetime": FakeData.fake_timestamp_io[1], "units": fd["unit"], "sampling_rate": None, "mapped_location_string": fd["mappedLocationString"], diff --git a/tests/test_file_writer.py b/tests/test_file_writer.py index db5ba63..db07bcb 100644 --- a/tests/test_file_writer.py +++ b/tests/test_file_writer.py @@ -80,13 +80,13 @@ def test_create_csv_low_freq( string_values=string_values, numeric_values=numeric_values, source_variable_id=variable_id, - observation_timestamp=observation_time.timestamp(), + observation_datetime=observation_time, units=u, mapped_location_string="mapped loc", csn=csn, mrn=mrn, ) - expected_line = f'"12345678","whatever","{variable_id}","","{u}","","1735726210.0","mapped loc","{expected_v_num}","{expected_v_str}"\n' + expected_line = f'"12345678","whatever","{variable_id}","","{u}","","2025-01-01T10:10:10+00:00","mapped loc","{expected_v_num}","{expected_v_str}"\n' expected_texts[u] += expected_line # need to check multiple files for multiple units for idx, ef in enumerate(expected_filenames): @@ -120,7 +120,7 @@ def test_create_csv_high_freq( numeric_values=[1, 2, 3.0], source_variable_id=variable_id, source_channel_id=channel_id, - observation_timestamp=observation_time.timestamp(), + observation_datetime=observation_time, units=units, sampling_rate=50, mapped_location_string="mapped loc", @@ -130,7 +130,7 @@ def test_create_csv_high_freq( expected_text = ( 'csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,numeric_values,string_values\n' - f'"12345678","whatever","{variable_id}","{channel_id or ""}","{units}","50","1735726210.0","mapped loc","[1, 2, 3.0]",""\n' + f'"12345678","whatever","{variable_id}","{channel_id or ""}","{units}","50","2025-01-01T10:10:10+00:00","mapped loc","[1, 2, 3.0]",""\n' ) _check_written_csv(expected_filename, expected_text) diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 9cc9f04..923894a 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -2,6 +2,7 @@ import os import re import shutil +from datetime import datetime, timezone from decimal import Decimal import pyarrow as pa @@ -75,6 +76,10 @@ def _numeric_rows_as_written(numeric_rows: list) -> list: return result +def _posix_to_datetime_utc(posix_time: float) -> datetime: + return datetime.fromtimestamp(posix_time, tz=timezone.utc) + + def _make_test_input_csv( monkeypatch, tmp_path, t: TestFileDescription ) -> tuple[list, list]: @@ -93,6 +98,7 @@ def _make_test_input_csv( source_channel_id = None if t.channel_id == "noCh" else t.channel_id row_time = t.start_timestamp for numeric_values, string_values in zip(numeric_rows, string_rows): + row_time_dt = _posix_to_datetime_utc(row_time) csv_writer.write_frame( numeric_values=( [float(v) for v in numeric_values] @@ -102,7 +108,7 @@ def _make_test_input_csv( string_values=string_values, source_variable_id=t.variable_id, source_channel_id=source_channel_id, - observation_timestamp=row_time, + observation_datetime=row_time_dt, units=t.units, sampling_rate=t.sampling_rate, mapped_location_string=t.location, @@ -248,28 +254,34 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): { "csn": file1.csn, "hashed_csn": file1.get_hashed_csn(), - "min_timestamp": file2.start_timestamp, - "max_timestamp": ( + "min_timestamp": _posix_to_datetime_utc( + file2.start_timestamp + ).isoformat(), + "max_timestamp": _posix_to_datetime_utc( file1.start_timestamp + file1.num_rows - 1 - ), # one sec per row + ).isoformat(), # one sec per row }, { "csn": file3.csn, "hashed_csn": file3.get_hashed_csn(), - "min_timestamp": file3.start_timestamp, - "max_timestamp": ( + "min_timestamp": _posix_to_datetime_utc( + file3.start_timestamp + ).isoformat(), + "max_timestamp": _posix_to_datetime_utc( file3.start_timestamp + file3.num_rows - 1 - ), # one sec per row + ).isoformat(), # one sec per row }, ], "2025-01-02": [ { "csn": file4.csn, "hashed_csn": file4.get_hashed_csn(), - "min_timestamp": file4.start_timestamp, - "max_timestamp": ( + "min_timestamp": _posix_to_datetime_utc( + file4.start_timestamp + ).isoformat(), + "max_timestamp": _posix_to_datetime_utc( file4.start_timestamp + file4.num_rows - 1 - ), # one sec per row + ).isoformat(), # one sec per row } ], } From 649a0188981eb8ad9bc3ee94dc10411c35f506bd Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Thu, 1 Oct 2026 11:36:41 +0100 Subject: [PATCH 2/4] explicit timespec Co-authored-by: Stephen Thompson --- src/csv_writer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/csv_writer.py b/src/csv_writer.py index cc085cb..a1ab041 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -89,7 +89,7 @@ def write_frame( source_channel_id if source_channel_id is not None else "", units, sampling_rate if sampling_rate is not None else "", - observation_datetime.isoformat(), + observation_datetime.isoformat(timespec="microseconds"), mapped_location_string, json.dumps(numeric_values) if numeric_values is not None else "", json.dumps(string_values) if string_values is not None else "", From 01699cdd4e45969a1f9392ebea6e43595235d66c Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Thu, 1 Oct 2026 11:44:12 +0100 Subject: [PATCH 3/4] Fix tests to match microsecond resolution --- tests/test_file_writer.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_file_writer.py b/tests/test_file_writer.py index db07bcb..3d0f593 100644 --- a/tests/test_file_writer.py +++ b/tests/test_file_writer.py @@ -86,7 +86,7 @@ def test_create_csv_low_freq( csn=csn, mrn=mrn, ) - expected_line = f'"12345678","whatever","{variable_id}","","{u}","","2025-01-01T10:10:10+00:00","mapped loc","{expected_v_num}","{expected_v_str}"\n' + expected_line = f'"12345678","whatever","{variable_id}","","{u}","","2025-01-01T10:10:10.000000+00:00","mapped loc","{expected_v_num}","{expected_v_str}"\n' expected_texts[u] += expected_line # need to check multiple files for multiple units for idx, ef in enumerate(expected_filenames): @@ -130,7 +130,7 @@ def test_create_csv_high_freq( expected_text = ( 'csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,numeric_values,string_values\n' - f'"12345678","whatever","{variable_id}","{channel_id or ""}","{units}","50","2025-01-01T10:10:10+00:00","mapped loc","[1, 2, 3.0]",""\n' + f'"12345678","whatever","{variable_id}","{channel_id or ""}","{units}","50","2025-01-01T10:10:10.000000+00:00","mapped loc","[1, 2, 3.0]",""\n' ) _check_written_csv(expected_filename, expected_text) From 25be712b73ae2a90bd72adaa95a3ed0b2cdf77c8 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Thu, 1 Oct 2026 11:41:44 +0100 Subject: [PATCH 4/4] Rename variable as per review comment --- src/controller.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/controller.py b/src/controller.py index 2bb4d85..2d1bfae 100644 --- a/src/controller.py +++ b/src/controller.py @@ -153,7 +153,7 @@ def outcome( try: location_string = message.get_mapped_location_string() - observation_timestamp = message.get_observation_time() + observation_datetime = message.get_observation_time() source_variable_id = message.get_source_variable_id() units = message.get_unit() mapped_location_string = message.get_mapped_location_string() @@ -206,14 +206,14 @@ def outcome( lookup_success = True try: matched_mrn = self.emap_db.get_matched_mrn( - location_string, observation_timestamp + location_string, observation_datetime ) except ValueError: lookup_success = False logger.error( "Ambiguous or non existent match for location %s, obs time %s", location_string, - observation_timestamp, + observation_datetime, exc_info=True, ) matched_mrn = ("unmatched_mrn", "unmatched_nhs", "unmatched_csn", False) @@ -230,7 +230,7 @@ def outcome( source_variable_id=source_variable_id, source_channel_id=source_channel_id, sampling_rate=sampling_rate, - observation_datetime=observation_timestamp, + observation_datetime=observation_datetime, units=units, mapped_location_string=mapped_location_string, csn=csn,