Use ISO/UTC timestamps in the Waveform CSV and parquet files

src/controller.py

@@ -4,7 +4,6 @@
 """
 
 from dataclasses import dataclass
-from datetime import datetime, timezone
 import logging
 from typing import Literal, Any, Optional
 
@@ -154,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()
@@ -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_datetime
             )
         except ValueError:
             lookup_success = False
             logger.error(
                 "Ambiguous or non existent match for location %s, obs time %s",
                 location_string,
-                observation_time,
+                observation_datetime,
                 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_datetime,
             units=units,
             mapped_location_string=mapped_location_string,
             csn=csn,

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(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 "",

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."""

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}")

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)

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

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

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"],

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.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):
@@ -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.000000+00:00","mapped loc","[1, 2, 3.0]",""\n'
     )
     _check_written_csv(expected_filename, expected_text)
 

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
             }
         ],
     }