From aaee72c158a596cc60fe2766ea33fd5bde73a760 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Thu, 13 Aug 2026 10:13:55 +0100 Subject: [PATCH 01/13] Document current Emap-Interchange ingress behaviour + link to HL7 replay doc --- README.md | 2 ++ docs/deployment.md | 3 ++- docs/overview.md | 12 ++++++++++++ 3 files changed, 16 insertions(+), 1 deletion(-) create mode 100644 docs/overview.md diff --git a/README.md b/README.md index cb9763f..606e82d 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ A controller for reading waveform data from a rabbitmq queue and processing it. +See [architecture overview diagram](https://github.com/SAFEHR-data/emap/blob/develop/docs/technical_overview/waveforms/pipeline.md) + # Running the Code ## Pre-reqs diff --git a/docs/deployment.md b/docs/deployment.md index 1c8940a..4a09d72 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -139,4 +139,5 @@ Waveform repo: `docker compose up -d` ### Replay old HL7 data -Not yet supported, see https://github.com/SAFEHR-data/emap/issues/139 +See [the Waveform docs in Emap](https://github.com/SAFEHR-data/emap/edit/develop/docs/dev/features/waveform_hf_data.md) +for how to replay old messages. diff --git a/docs/overview.md b/docs/overview.md new file mode 100644 index 0000000..a1d21bf --- /dev/null +++ b/docs/overview.md @@ -0,0 +1,12 @@ +# Reading from the rabbitmq queue + +The waveform-controller container will process messages on a rabbitmq queue that +are JSON representations of the Emap-Interchange message types +`WaveformMessage` and, imminently, `WaveformLowFreqMessage`. + +As you can see from `test_controller.test_controller_callback`, the following actions are +expected in various error cases: + +* **Bad data (eg. missing column)**: REJECT without requeue, because it's assumed the message will never work. +* **Postgress connection failed**: REJECT with requeue, because failure is assumed to be unrelated to the received message. +* **Patient has research opt-out set** REJECT without requeue From 3532927d0b4b9821d397e1b55cb61660b669394c Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Thu, 13 Aug 2026 14:36:41 +0100 Subject: [PATCH 02/13] Update fake data to be more realistic, check the happy path a bit more carefully, and to have a HF and LF version. (tests with LF currently fail because it's not implemented yet) --- tests/test_controller.py | 119 ++++++++++++++++++++++++++++++++++----- 1 file changed, 104 insertions(+), 15 deletions(-) diff --git a/tests/test_controller.py b/tests/test_controller.py index d4fd611..709d0f4 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -7,6 +7,103 @@ from controller import WaveformController +class FakeData: + def __init__(self, bad_data): + self.bad_data = bad_data + self.fake_data = FakeData._base_fake_data() + + @staticmethod + def _base_fake_data() -> dict: + return { + "sourceSystem": None, + "sourceMessageId": "UCHT03ICURM09_t20240912080130_00003_1_10", + "sourceLocationString": "foo", + "sourceObservationType": "waveform", + "mappedVariableDescription": "P0.1 Occlusion Pressure", + "mappedLocationString": "loc", + "observationTime": datetime.now().timestamp(), + "sourceVariableId": "27", + "unit": "uV", + } + + +class FakeHFData(FakeData): + def get_fake_data(self) -> dict: + self.fake_data["@class"] = ( + "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformMessage" + ) + # simulate a missing key + if not self.bad_data: + self.fake_data["sourceChannelId"] = "1" + self.fake_data["samplingRate"] = 50 + self.fake_data["numericValues"] = { + "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", + "value": [956.793, 945.615], + "status": "SAVE", + } + return self.fake_data + + def get_expected_write_frame_args(self): + """If not bad data, what should write_frame be called with?""" + fd = self.fake_data + return ( + fd["numericValues"], + fd["sourceVariableId"], + fd["sourceChannelId"], + fd["observationTime"], + fd["unit"], + fd["samplingRate"], + fd["mappedLocationString"], + "csn", + "mrn", + ) + + +class FakeLFData(FakeData): + def get_fake_data(self) -> dict: + self.fake_data["@class"] = ( + "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformLowFreqMessage" + ) + self.fake_data["sourceValue"] = { + "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", + "value": "0.8", + "status": "SAVE", + } + self.fake_data["numericValue"] = { + "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", + "value": 0.8, + "status": "SAVE", + } + self.fake_data["stringValue"] = { + "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", + "value": None, + "status": "IGNORE", + } + if self.bad_data: + # simulate a missing key + del self.fake_data["sourceVariableId"] + return self.fake_data + + def get_expected_write_frame_args(self): + """If not bad data, what should write_frame be called with?""" + fd = self.fake_data + return ( + fd["numericValues"], + fd["sourceVariableId"], + fd["sourceChannelId"], + fd["observationTime"], + fd["unit"], + fd["samplingRate"], + fd["mappedLocationString"], + "csn", + "mrn", + ) + + +@pytest.mark.parametrize( + "fake_data_class", + [FakeHFData, FakeLFData], +) @pytest.mark.parametrize( "opt_out", [True, False], @@ -19,7 +116,9 @@ "bad_data", [True, False], ) -def test_controller_callback(monkeypatch, opt_out, db_connect_failure, bad_data): +def test_controller_callback( + monkeypatch, fake_data_class, opt_out, db_connect_failure, bad_data +): emap_db_mock = Mock() if db_connect_failure: emap_db_mock.get_row.side_effect = ConnectionError("mock database error") @@ -30,19 +129,8 @@ def test_controller_callback(monkeypatch, opt_out, db_connect_failure, bad_data) write_frame_mock = Mock(return_value=True) monkeypatch.setattr("controller.writer.write_frame", write_frame_mock) - fake_data = { - "sourceLocationString": "foo", - "mappedLocationString": "loc", - "observationTime": datetime.now().timestamp(), - "sourceVariableId": "27", - "sourceChannelId": "1", - "samplingRate": 50, - "unit": "uV", - "numericValues": "[1,2,3]", - } - if bad_data: - # simulate a missing key - del fake_data["sourceChannelId"] + fake_data_obj = fake_data_class(bad_data=bad_data) + fake_data = fake_data_obj.get_fake_data() fake_data_str = json.dumps(fake_data) controller = WaveformController() @@ -76,6 +164,7 @@ def test_controller_callback(monkeypatch, opt_out, db_connect_failure, bad_data) channel_mock.basic_ack.assert_not_called() else: # happy path - write_frame_mock.assert_called_once() + expected_write_frame_args = fake_data_obj.get_expected_write_frame_args() + write_frame_mock.assert_called_once_with(*expected_write_frame_args) channel_mock.basic_reject.assert_not_called() channel_mock.basic_ack.assert_called_once_with(delivery_tag) From accde07ed9efd1814e11ed401e9f770ab78491d4 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Fri, 14 Aug 2026 10:43:46 +0100 Subject: [PATCH 03/13] Understand the different types of incoming waveform interchange message. Define the data that needs to be passed for output. --- src/controller.py | 54 +++++++++++------- src/csv_writer.py | 23 +++++++- src/emap_interchange/__init__.py | 0 src/emap_interchange/messages.py | 97 ++++++++++++++++++++++++++++++++ tests/test_controller.py | 82 +++++++++++++++------------ tests/test_file_writer.py | 18 +++--- 6 files changed, 206 insertions(+), 68 deletions(-) create mode 100644 src/emap_interchange/__init__.py create mode 100644 src/emap_interchange/messages.py diff --git a/src/controller.py b/src/controller.py index 3dc3099..8ee875b 100644 --- a/src/controller.py +++ b/src/controller.py @@ -3,13 +3,17 @@ based on https://www.rabbitmq.com/tutorials/tutorial-one-python """ -import json from datetime import datetime, timezone import logging import pika import db as db # type:ignore import settings as settings # type:ignore import csv_writer as writer # type:ignore +from emap_interchange.messages import ( + WaveformBaseMessage, + WaveformHighFreqMessage, + WaveformLowFreqMessage, +) logging.basicConfig(format="%(levelname)s:%(asctime)s: %(message)s") logger = logging.getLogger(__name__) @@ -48,16 +52,28 @@ def __init__(self): def waveform_callback(self, ch, method_frame, _header_frame, body): logger.debug("Message received of length %s", len(body)) - data = json.loads(body) try: - location_string = data["mappedLocationString"] - observation_timestamp = data["observationTime"] - source_variable_id = data["sourceVariableId"] - source_channel_id = data["sourceChannelId"] - sampling_rate = data["samplingRate"] - units = data["unit"] - waveform_data = data["numericValues"] - mapped_location_string = data["mappedLocationString"] + message = WaveformBaseMessage.from_json(body) + except TypeError as e: + logger.error("Skipping, could not understand message type %s", e) + reject_message(ch, method_frame.delivery_tag, False) + return + + try: + data_kwarg = {} + location_string = message.get_mapped_location_string() + observation_timestamp = message.get_observation_time() + source_variable_id = message.get_source_variable_id() + units = message.get_unit() + mapped_location_string = message.get_mapped_location_string() + if isinstance(message, WaveformHighFreqMessage): + sampling_rate = message.get_sampling_rate() + source_channel_id = message.get_source_channel_id() + waveform_data = message.get_numeric_values() + data_kwarg["waveform_data"] = waveform_data + elif isinstance(message, WaveformLowFreqMessage): + data_kwarg["single_value_str"] = message.get_string_value() + data_kwarg["single_value_numeric"] = message.get_numeric_value() logger.debug( "Message is for loc %s, var %s, ch %s", location_string, @@ -98,15 +114,15 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): return if writer.write_frame( - waveform_data, - source_variable_id, - source_channel_id, - observation_timestamp, - units, - sampling_rate, - mapped_location_string, - csn, - mrn, + source_variable_id=source_variable_id, + source_channel_id=source_channel_id, + observation_timestamp=observation_timestamp, + units=units, + sampling_rate=sampling_rate, + mapped_location_string=mapped_location_string, + csn=csn, + mrn=mrn, + **data_kwarg, ): if lookup_success: ack_message(ch, method_frame.delivery_tag) diff --git a/src/csv_writer.py b/src/csv_writer.py index 66948b8..da28977 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -2,6 +2,7 @@ import csv from datetime import datetime +from typing import Optional from locations import WAVEFORM_ORIGINAL_CSV, make_file_name, FILE_STEM_PATTERN @@ -30,7 +31,10 @@ def create_file_name( def write_frame( - waveform_data: dict, + *, + waveform_data: Optional[str] = None, + single_value_str: Optional[str] = None, + single_value_numeric: Optional[float] = None, source_variable_id: str, source_channel_id: str, observation_timestamp: float, @@ -40,10 +44,24 @@ def write_frame( csn: str, mrn: str, ) -> bool: - """Appends a frame of waveform data to a csv file (creates file if it doesn't exist. + """Appends a frame of waveform data to a csv file (creates file if it doesn't + exist). Exactly ONE of waveform_data, single_value_str, single_value_numeric must + contain a non-None value. :param waveform_data: The waveform data to write, as a + JSON value string of comma-separated numerics. Ie. as it comes straight out of the + interchange message. :single_value_str: The single value string of the waveform + data. :single_value_numeric: The single value string of the waveform data. :return: True if write was successful. """ + num_non_nones = sum( + 1 + for x in (waveform_data, single_value_str, single_value_numeric) + if x is not None + ) + if num_non_nones != 1: + raise ValueError( + "Exactly ONE of waveform_data, single_value_str, single_value_numeric must be not None" + ) observation_datetime = datetime.fromtimestamp(observation_timestamp) filename = WAVEFORM_ORIGINAL_CSV / create_file_name( @@ -60,7 +78,6 @@ def write_frame( with open(filename, "a") as fileout: wv_writer = csv.writer(fileout, delimiter=",") - waveform_data = waveform_data.get("value", "") wv_writer.writerow( [ diff --git a/src/emap_interchange/__init__.py b/src/emap_interchange/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/emap_interchange/messages.py b/src/emap_interchange/messages.py new file mode 100644 index 0000000..3163cdb --- /dev/null +++ b/src/emap_interchange/messages.py @@ -0,0 +1,97 @@ +# Python version of interchange messages +# Ideally this would be generated automatically by the Java +# code as part of its build process, but for now it's just copied +# and modified +import json + + +class WaveformBaseMessage: + def __init__(self, data): + self.data = data + + @staticmethod + def from_json(json_data) -> "WaveformBaseMessage": + data = json.loads(json_data) + message_type = data.get("@class") + message_cls: type + if ( + message_type + == "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformHighFreqMessage" + ): + message_cls = WaveformHighFreqMessage + elif ( + message_type + == "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformLowFreqMessage" + ): + message_cls = WaveformLowFreqMessage + else: + raise TypeError("Unknown message type {}".format(message_type)) + + return message_cls(data) + + def get_observation_time(self): + """Time of the observation.""" + return self.data["observationTime"] + + def get_source_location_string(self): + """Location string according to the original data source.""" + return self.data["sourceLocationString"] + + def get_mapped_location_string(self): + """Location string, mapped by the data source to the canonical Emap format, + which matches what we get from the main HL7 ADT feed.""" + return self.data["mappedLocationString"] + + def get_source_observation_type(self): + """Do we want to be more specific here? + + Eg. carescape, etc Eg. get from the CSV metadata and prefix with "waveform-" + """ + return self.data["sourceObservationType"] + + def get_source_variable_id(self): + """Variable ID according to the source system. + + Has previously been referred to as stream ID, so you may see that in some + places. + """ + return self.data["sourceVariableId"] + + def get_mapped_variable_description(self): + """Variable (aka stream) description mapped by the data source.""" + return self.data["mappedVariableDescription"] + + def get_unit(self): + """Unit of the measurement.""" + return self.data["unit"] + + +class WaveformHighFreqMessage(WaveformBaseMessage): + def get_source_channel_id(self): + """Channel ID according to the source system.""" + return self.data["sourceChannelId"] + + def get_sampling_rate(self): + """Sampling rate in Hz.""" + return self.data["samplingRate"] + + def get_numeric_values(self): + """Numeric array.""" + return self.data["numericValues"]["value"] + + +class WaveformLowFreqMessage(WaveformBaseMessage): + def get_source_value(self): + """Unmapped value.""" + return self.data["sourceValue"]["value"] + + def get_numeric_value(self): + """Mapped value, if it's a numerical value.""" + return self.data["numericValue"]["value"] + + def get_string_value(self): + """Mapped value, if it's a string. + + Also use for categorical, eg. "Flow Trig" + """ + return self.data["stringValue"]["value"] diff --git a/tests/test_controller.py b/tests/test_controller.py index 709d0f4..7fe8423 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -8,8 +8,8 @@ class FakeData: - def __init__(self, bad_data): - self.bad_data = bad_data + def __init__(self, missing_key): + self.missing_key = missing_key self.fake_data = FakeData._base_fake_data() @staticmethod @@ -30,10 +30,10 @@ def _base_fake_data() -> dict: class FakeHFData(FakeData): def get_fake_data(self) -> dict: self.fake_data["@class"] = ( - "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformMessage" + "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformHighFreqMessage" ) # simulate a missing key - if not self.bad_data: + if not self.missing_key: self.fake_data["sourceChannelId"] = "1" self.fake_data["samplingRate"] = 50 self.fake_data["numericValues"] = { @@ -43,20 +43,20 @@ def get_fake_data(self) -> dict: } return self.fake_data - def get_expected_write_frame_args(self): + def get_expected_write_frame_kwargs(self): """If not bad data, what should write_frame be called with?""" fd = self.fake_data - return ( - fd["numericValues"], - fd["sourceVariableId"], - fd["sourceChannelId"], - fd["observationTime"], - fd["unit"], - fd["samplingRate"], - fd["mappedLocationString"], - "csn", - "mrn", - ) + return { + "waveform_data": fd["numericValues"]["value"], + "source_variable_id": fd["sourceVariableId"], + "source_channel_id": fd["sourceChannelId"], + "observation_timestamp": fd["observationTime"], + "units": fd["unit"], + "sampling_rate": fd["samplingRate"], + "mapped_location_string": fd["mappedLocationString"], + "csn": "csn", + "mrn": "mrn", + } class FakeLFData(FakeData): @@ -79,25 +79,25 @@ def get_fake_data(self) -> dict: "value": None, "status": "IGNORE", } - if self.bad_data: + if self.missing_key: # simulate a missing key del self.fake_data["sourceVariableId"] return self.fake_data - def get_expected_write_frame_args(self): + def get_expected_write_frame_kwargs(self): """If not bad data, what should write_frame be called with?""" fd = self.fake_data - return ( - fd["numericValues"], - fd["sourceVariableId"], - fd["sourceChannelId"], - fd["observationTime"], - fd["unit"], - fd["samplingRate"], - fd["mappedLocationString"], - "csn", - "mrn", - ) + return { + "waveform_data": fd["numericValues"]["value"], + "source_variable_id": fd["sourceVariableId"], + "source_channel_id": fd["sourceChannelId"], + "observation_timestamp": fd["observationTime"], + "units": fd["unit"], + "sampling_rate": fd["samplingRate"], + "mapped_location_string": fd["mappedLocationString"], + "csn": "csn", + "mrn": "mrn", + } @pytest.mark.parametrize( @@ -113,11 +113,12 @@ def get_expected_write_frame_args(self): [True, False], ) @pytest.mark.parametrize( - "bad_data", - [True, False], + "bad_data_type", + # 0 is not bad data, 1,2,3 are various different kinds of bad data + range(4), ) def test_controller_callback( - monkeypatch, fake_data_class, opt_out, db_connect_failure, bad_data + monkeypatch, fake_data_class, opt_out, db_connect_failure, bad_data_type ): emap_db_mock = Mock() if db_connect_failure: @@ -129,8 +130,15 @@ def test_controller_callback( write_frame_mock = Mock(return_value=True) monkeypatch.setattr("controller.writer.write_frame", write_frame_mock) - fake_data_obj = fake_data_class(bad_data=bad_data) + # Simulate various kinds of bad data. Make sure to keep the range parameter + # bad_data_type up to date with the number of possible failures + fake_data_obj = fake_data_class(missing_key=(bad_data_type == 1)) fake_data = fake_data_obj.get_fake_data() + match bad_data_type: + case 2: + del fake_data["@class"] + case 3: + fake_data["@class"] = fake_data["@class"].replace("e", "x") fake_data_str = json.dumps(fake_data) controller = WaveformController() @@ -142,11 +150,11 @@ def test_controller_callback( controller.waveform_callback(channel_mock, method_frame_mock, None, fake_data_str) - if not bad_data: + if not bad_data_type: # we at least tried to query the DB emap_db_mock.get_row.assert_called_once() - if bad_data: + if bad_data_type: write_frame_mock.assert_not_called() # db should not even have been queried if data was bad emap_db_mock.get_row.assert_not_called() @@ -164,7 +172,7 @@ def test_controller_callback( channel_mock.basic_ack.assert_not_called() else: # happy path - expected_write_frame_args = fake_data_obj.get_expected_write_frame_args() - write_frame_mock.assert_called_once_with(*expected_write_frame_args) + expected_write_frame_kwargs = fake_data_obj.get_expected_write_frame_kwargs() + write_frame_mock.assert_called_once_with(**expected_write_frame_kwargs) channel_mock.basic_reject.assert_not_called() channel_mock.basic_ack.assert_called_once_with(delivery_tag) diff --git a/tests/test_file_writer.py b/tests/test_file_writer.py index 6bb6983..0616822 100644 --- a/tests/test_file_writer.py +++ b/tests/test_file_writer.py @@ -33,15 +33,15 @@ def test_create_file_name_handles_units( mrn = "whatever" csv_writer.write_frame( - {"value": "[1,2,3]"}, - variable_id, - channel_id, - observation_time.timestamp(), - units, - 50, - "mapped loc", - csn, - mrn, + waveform_data="[1,2,3]", + source_variable_id=variable_id, + source_channel_id=channel_id, + observation_timestamp=observation_time.timestamp(), + units=units, + sampling_rate=50, + mapped_location_string="mapped loc", + csn=csn, + mrn=mrn, ) # check that we can find the data again in its expected place From 6b3aed8bb41e63b780d45422d59285b62b7f5323 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Fri, 14 Aug 2026 16:57:30 +0100 Subject: [PATCH 04/13] Check the CSV file contents. Implement writing low- or high- freq CSVs. --- src/controller.py | 39 ++++++++---- src/csv_writer.py | 62 +++++++++++------- tests/test_controller.py | 80 +++++++++++++++++------- tests/test_file_writer.py | 128 ++++++++++++++++++++++++++++++++++---- 4 files changed, 241 insertions(+), 68 deletions(-) diff --git a/src/controller.py b/src/controller.py index 8ee875b..14f85a9 100644 --- a/src/controller.py +++ b/src/controller.py @@ -67,19 +67,34 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): units = message.get_unit() mapped_location_string = message.get_mapped_location_string() if isinstance(message, WaveformHighFreqMessage): - sampling_rate = message.get_sampling_rate() - source_channel_id = message.get_source_channel_id() + data_kwarg["sampling_rate"] = message.get_sampling_rate() + data_kwarg["source_channel_id"] = message.get_source_channel_id() waveform_data = message.get_numeric_values() - data_kwarg["waveform_data"] = waveform_data + data_kwarg["values"] = waveform_data + logger.debug( + "WaveformHighFreqMessage is for loc %s, var %s, ch %s", + location_string, + source_variable_id, + data_kwarg["source_channel_id"], + ) elif isinstance(message, WaveformLowFreqMessage): - data_kwarg["single_value_str"] = message.get_string_value() - data_kwarg["single_value_numeric"] = message.get_numeric_value() - logger.debug( - "Message is for loc %s, var %s, ch %s", - location_string, - source_variable_id, - source_channel_id, - ) + string_value = message.get_string_value() + if string_value is not None: + data_kwarg["string_value"] = string_value + + numeric_value = message.get_numeric_value() + if numeric_value is not None: + data_kwarg["numeric_value"] = numeric_value + + logger.debug( + "WaveformLowFreqMessage is for loc %s, var %s", + location_string, + source_variable_id, + ) + else: + raise RuntimeError( + "Unrecognized message type but should have dealt with this by now?" + ) except KeyError as e: reject_message(ch, method_frame.delivery_tag, False) logger.error( @@ -115,10 +130,8 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): if writer.write_frame( source_variable_id=source_variable_id, - source_channel_id=source_channel_id, observation_timestamp=observation_timestamp, units=units, - sampling_rate=sampling_rate, mapped_location_string=mapped_location_string, csn=csn, mrn=mrn, diff --git a/src/csv_writer.py b/src/csv_writer.py index da28977..2aea694 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -9,7 +9,7 @@ def create_file_name( source_variable_id: str, - source_channel_id: str, + source_channel_id: Optional[str], observation_time: datetime, csn: str, units: str, @@ -32,35 +32,33 @@ def create_file_name( def write_frame( *, - waveform_data: Optional[str] = None, - single_value_str: Optional[str] = None, - single_value_numeric: Optional[float] = None, + values: Optional[str] = None, + string_value: Optional[str] = None, + numeric_value: Optional[float] = None, source_variable_id: str, - source_channel_id: str, + source_channel_id: Optional[str] = None, observation_timestamp: float, units: str, - sampling_rate: int, + sampling_rate: Optional[int] = None, mapped_location_string: str, csn: str, mrn: str, ) -> bool: """Appends a frame of waveform data to a csv file (creates file if it doesn't - exist). Exactly ONE of waveform_data, single_value_str, single_value_numeric must - contain a non-None value. :param waveform_data: The waveform data to write, as a - JSON value string of comma-separated numerics. Ie. as it comes straight out of the - interchange message. :single_value_str: The single value string of the waveform - data. :single_value_numeric: The single value string of the waveform data. + exist). Exactly ONE of values, string_value, numeric_value must contain a non-None + value. :param values: The waveform data to write, as a JSON value string of comma- + separated numerics. Ie. as it comes straight out of the interchange message. + :string_value: The single value string of the waveform data. :numeric_value: The + single value string of the waveform data. :return: True if write was successful. """ num_non_nones = sum( - 1 - for x in (waveform_data, single_value_str, single_value_numeric) - if x is not None + 1 for x in (values, string_value, numeric_value) if x is not None ) if num_non_nones != 1: raise ValueError( - "Exactly ONE of waveform_data, single_value_str, single_value_numeric must be not None" + "Exactly ONE of values, string_value, numeric_value must be not None" ) observation_datetime = datetime.fromtimestamp(observation_timestamp) @@ -69,18 +67,24 @@ def write_frame( ) filename.parent.mkdir(exist_ok=True, parents=True) + if values is not None: + # HF + csv_header = "csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,values\n" + else: + # LF + csv_header = "csn,mrn,source_variable_id,units,timestamp,location,string_value,numeric_value\n" # write header if is new file if not filename.exists(): with open(filename, "w") as fileout: - fileout.write( - "csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,values\n" - ) + fileout.write(csv_header) with open(filename, "a") as fileout: - wv_writer = csv.writer(fileout, delimiter=",") + # predictable quoting makes testing easier + wv_writer = csv.writer(fileout, delimiter=",", quoting=csv.QUOTE_ALL) - wv_writer.writerow( - [ + if values is not None: + # HF + row_array = [ csn, mrn, source_variable_id, @@ -89,8 +93,20 @@ def write_frame( sampling_rate, observation_timestamp, mapped_location_string, - waveform_data, + values, ] - ) + else: + row_array = [ + csn, + mrn, + source_variable_id, + units, + observation_timestamp, + mapped_location_string, + string_value if string_value is not None else "", + numeric_value if numeric_value is not None else "", + ] + + wv_writer.writerow(row_array) return True diff --git a/tests/test_controller.py b/tests/test_controller.py index 7fe8423..2c4acd3 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1,3 +1,4 @@ +import copy import json from datetime import datetime from unittest.mock import Mock @@ -8,9 +9,13 @@ class FakeData: - def __init__(self, missing_key): + """Fake data to be used for building an Emap-Interchange JSON string for testing + purposes.""" + + def __init__(self, missing_key, value_type): self.missing_key = missing_key self.fake_data = FakeData._base_fake_data() + self.value_type = value_type @staticmethod def _base_fake_data() -> dict: @@ -43,11 +48,11 @@ def get_fake_data(self) -> dict: } return self.fake_data - def get_expected_write_frame_kwargs(self): + def get_expected_write_frame_kwargs(self) -> dict: """If not bad data, what should write_frame be called with?""" fd = self.fake_data return { - "waveform_data": fd["numericValues"]["value"], + "values": fd["numericValues"]["value"], "source_variable_id": fd["sourceVariableId"], "source_channel_id": fd["sourceChannelId"], "observation_timestamp": fd["observationTime"], @@ -64,42 +69,68 @@ def get_fake_data(self) -> dict: self.fake_data["@class"] = ( "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformLowFreqMessage" ) - self.fake_data["sourceValue"] = { + + ignore_val = { "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", - "value": "0.8", - "status": "SAVE", + "value": None, + "status": "IGNORE", } - self.fake_data["numericValue"] = { + if self.value_type == "numeric": + source_val = "0.8" + self.fake_data["numericValue"] = { + "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", + "value": 0.8, + "status": "SAVE", + } + self.fake_data["stringValue"] = copy.copy(ignore_val) + elif self.value_type == "string": + source_val = "some categorical" + self.fake_data["stringValue"] = { + "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", + "value": source_val, + "status": "SAVE", + } + self.fake_data["numericValue"] = copy.copy(ignore_val) + else: + raise ValueError("must be numeric or string") + self.fake_data["sourceValue"] = { "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", - "value": 0.8, + "value": source_val, "status": "SAVE", } - self.fake_data["stringValue"] = { - "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", - "value": None, - "status": "IGNORE", - } if self.missing_key: # simulate a missing key del self.fake_data["sourceVariableId"] return self.fake_data - def get_expected_write_frame_kwargs(self): - """If not bad data, what should write_frame be called with?""" + def get_expected_write_frame_kwargs(self) -> dict: + """If not bad data, what should write_frame be called with? + + Keys: CSV file column names + Values: using interchange field names + """ fd = self.fake_data - return { - "waveform_data": fd["numericValues"]["value"], + expected = { "source_variable_id": fd["sourceVariableId"], - "source_channel_id": fd["sourceChannelId"], "observation_timestamp": fd["observationTime"], "units": fd["unit"], - "sampling_rate": fd["samplingRate"], "mapped_location_string": fd["mappedLocationString"], "csn": "csn", "mrn": "mrn", } + if self.value_type == "numeric": + expected["numeric_value"] = fd["numericValue"]["value"] + elif self.value_type == "string": + expected["string_value"] = fd["stringValue"]["value"] + return expected + +@pytest.mark.parametrize( + # only affect LF tests so is redundant for HF + "lf_value_type", + ["string", "numeric"], +) @pytest.mark.parametrize( "fake_data_class", [FakeHFData, FakeLFData], @@ -118,7 +149,12 @@ def get_expected_write_frame_kwargs(self): range(4), ) def test_controller_callback( - monkeypatch, fake_data_class, opt_out, db_connect_failure, bad_data_type + monkeypatch, + lf_value_type, + fake_data_class, + opt_out, + db_connect_failure, + bad_data_type, ): emap_db_mock = Mock() if db_connect_failure: @@ -132,7 +168,9 @@ def test_controller_callback( # Simulate various kinds of bad data. Make sure to keep the range parameter # bad_data_type up to date with the number of possible failures - fake_data_obj = fake_data_class(missing_key=(bad_data_type == 1)) + fake_data_obj = fake_data_class( + missing_key=(bad_data_type == 1), value_type=lf_value_type + ) fake_data = fake_data_obj.get_fake_data() match bad_data_type: case 2: diff --git a/tests/test_file_writer.py b/tests/test_file_writer.py index 0616822..de0ad81 100644 --- a/tests/test_file_writer.py +++ b/tests/test_file_writer.py @@ -1,4 +1,5 @@ import os +from typing import Optional import pytest @@ -7,6 +8,92 @@ import locations +@pytest.mark.parametrize( + "units, variable_id, values, expected_filenames", + [ + # categorical, will have been mapped to string + ( + ["unitless"], + "584", + ["Pressure Support / CPAP (PS)"], + ["2025-01-01/2025-01-01.12345678.584.noCh.unitless.csv"], + ), + # string value + ( + ["unitless"], + "1190", + ["1:2"], + ["2025-01-01/2025-01-01.12345678.1190.noCh.unitless.csv"], + ), + # numerical, and also a variable with more than one unit at the same time + ( + ["%", "s"], + "1408", + [5, 0.2], + # units should probably be removed from the filename altogether, + # as it will create a separate file for each + [ + "2025-01-01/2025-01-01.12345678.1408.noCh.percent.csv", + "2025-01-01/2025-01-01.12345678.1408.noCh.s.csv", + ], + ), + ], +) +def test_create_csv_low_freq( + monkeypatch, + tmp_path, + units: list[str], + variable_id: str, + values: list, + expected_filenames, +): + """ + :param values: one value per line to write! + """ + # check test is valid + assert len(units) == len(values) == len(expected_filenames) + + _setup_write_csv(monkeypatch, tmp_path) + + observation_time = datetime(2025, 1, 1, 10, 10, 10, tzinfo=timezone.utc) + csn = "12345678" + mrn = "whatever" + + # it creates a separate file for each unit, so this becomes quite messy + expected_header = "csn,mrn,source_variable_id,units,timestamp,location,string_value,numeric_value\n" + expected_texts = dict.fromkeys(units, expected_header) + for idx, u in enumerate(units): + v = values[idx] + if type(v) is str: + value_kwarg = {"string_value": v} + expected_v_str = v + expected_v_num = "" + elif type(v) is float or type(v) is int: + value_kwarg = {"numeric_value": v} + expected_v_str = "" + expected_v_num = str(v) + else: + raise ValueError(v) + + csv_writer.write_frame( + **value_kwarg, + source_variable_id=variable_id, + observation_timestamp=observation_time.timestamp(), + units=u, + mapped_location_string="mapped loc", + csn=csn, + mrn=mrn, + ) + expected_line = ( + f'"12345678","whatever","{variable_id}","{u}","1735726210.0",' + f'"mapped loc","{expected_v_str}","{expected_v_num}"\n' + ) + expected_texts[u] += expected_line + # need to check all files now + for idx, ef in enumerate(expected_filenames): + _check_write_csv(ef, expected_texts[units[idx]]) + + @pytest.mark.parametrize( "units, variable_id, channel_id, expected_filename", [ @@ -16,24 +103,22 @@ ("%", "11", "3", "2025-01-01/2025-01-01.12345678.11.3.percent.csv"), ], ) -def test_create_file_name_handles_units( - monkeypatch, units, variable_id, channel_id, expected_filename, tmp_path +def test_create_csv_high_freq( + monkeypatch, + units: str, + variable_id: str, + channel_id: Optional[str], + expected_filename: str, + tmp_path, ): - # treat the normal absolute path as if it were a relative path, so we can put - # a prefix on it (this code is usually run in a container) - original_csv_dir = tmp_path / locations.WAVEFORM_ORIGINAL_CSV.relative_to("/") - monkeypatch.setattr(csv_writer, "WAVEFORM_ORIGINAL_CSV", original_csv_dir) - monkeypatch.setattr(locations, "WAVEFORM_ORIGINAL_CSV", original_csv_dir) - - # the only precondition is that the base dir must exist - original_csv_dir.parent.mkdir(parents=True, exist_ok=True) + _setup_write_csv(monkeypatch, tmp_path) observation_time = datetime(2025, 1, 1, 10, 10, 10, tzinfo=timezone.utc) csn = "12345678" mrn = "whatever" csv_writer.write_frame( - waveform_data="[1,2,3]", + values="[1,2,3]", source_variable_id=variable_id, source_channel_id=channel_id, observation_timestamp=observation_time.timestamp(), @@ -44,6 +129,27 @@ def test_create_file_name_handles_units( mrn=mrn, ) + expected_text = ( + 'csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,values\n' + f'"12345678","whatever","{variable_id}","{channel_id or ""}","{units}","50","1735726210.0","mapped loc","[1,2,3]"\n' + ) + _check_write_csv(expected_filename, expected_text) + + +def _setup_write_csv(monkeypatch, tmp_path): + # treat the normal absolute path as if it were a relative path, so we can put + # a prefix on it (this code is usually run in a container) + original_csv_dir = tmp_path / locations.WAVEFORM_ORIGINAL_CSV.relative_to("/") + monkeypatch.setattr(csv_writer, "WAVEFORM_ORIGINAL_CSV", original_csv_dir) + monkeypatch.setattr(locations, "WAVEFORM_ORIGINAL_CSV", original_csv_dir) + + # the only precondition is that the base dir must exist + original_csv_dir.parent.mkdir(parents=True, exist_ok=True) + + +def _check_write_csv(expected_filename, expected_text): # check that we can find the data again in its expected place expected_csv_path = locations.WAVEFORM_ORIGINAL_CSV / expected_filename assert os.path.exists(expected_csv_path) + actual_text = expected_csv_path.read_text() + assert actual_text == expected_text From 124d37f88ecd9e50f37705394f7e47cb73cc3cd8 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Sun, 16 Aug 2026 21:33:34 +0100 Subject: [PATCH 05/13] Simplify the CSV output format so that it's the same between LF and HF, otherwise parquet conversion is going to be too complex --- src/controller.py | 26 ++++++++----- src/csv_writer.py | 63 +++++++++++--------------------- src/emap_interchange/messages.py | 6 +-- tests/test_controller.py | 18 ++++++--- tests/test_file_writer.py | 35 +++++++++--------- 5 files changed, 71 insertions(+), 77 deletions(-) diff --git a/src/controller.py b/src/controller.py index 14f85a9..e54dc98 100644 --- a/src/controller.py +++ b/src/controller.py @@ -5,6 +5,7 @@ from datetime import datetime, timezone import logging + import pika import db as db # type:ignore import settings as settings # type:ignore @@ -60,31 +61,35 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): return try: - data_kwarg = {} location_string = message.get_mapped_location_string() observation_timestamp = message.get_observation_time() source_variable_id = message.get_source_variable_id() units = message.get_unit() mapped_location_string = message.get_mapped_location_string() + source_channel_id = None + sampling_rate = None + numeric_values = None + string_values = None if isinstance(message, WaveformHighFreqMessage): - data_kwarg["sampling_rate"] = message.get_sampling_rate() - data_kwarg["source_channel_id"] = message.get_source_channel_id() - waveform_data = message.get_numeric_values() - data_kwarg["values"] = waveform_data + sampling_rate = message.get_sampling_rate() + source_channel_id = message.get_source_channel_id() + numeric_values = message.get_numeric_values() logger.debug( "WaveformHighFreqMessage is for loc %s, var %s, ch %s", location_string, source_variable_id, - data_kwarg["source_channel_id"], + source_channel_id, ) elif isinstance(message, WaveformLowFreqMessage): + # Wrap single values in arrays so they can go in the same + # CSV (and parquet...) columns as for HF string_value = message.get_string_value() if string_value is not None: - data_kwarg["string_value"] = string_value + string_values = [string_value] numeric_value = message.get_numeric_value() if numeric_value is not None: - data_kwarg["numeric_value"] = numeric_value + numeric_values = [numeric_value] logger.debug( "WaveformLowFreqMessage is for loc %s, var %s", @@ -130,12 +135,15 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): if writer.write_frame( source_variable_id=source_variable_id, + source_channel_id=source_channel_id, + sampling_rate=sampling_rate, observation_timestamp=observation_timestamp, units=units, mapped_location_string=mapped_location_string, csn=csn, mrn=mrn, - **data_kwarg, + numeric_values=numeric_values, + string_values=string_values, ): if lookup_success: ack_message(ch, method_frame.delivery_tag) diff --git a/src/csv_writer.py b/src/csv_writer.py index 2aea694..96aacdf 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -32,9 +32,8 @@ def create_file_name( def write_frame( *, - values: Optional[str] = None, - string_value: Optional[str] = None, - numeric_value: Optional[float] = None, + numeric_values: Optional[list[float]] = None, + string_values: Optional[list[str]] = None, source_variable_id: str, source_channel_id: Optional[str] = None, observation_timestamp: float, @@ -45,20 +44,14 @@ def write_frame( mrn: str, ) -> bool: """Appends a frame of waveform data to a csv file (creates file if it doesn't - exist). Exactly ONE of values, string_value, numeric_value must contain a non-None - value. :param values: The waveform data to write, as a JSON value string of comma- - separated numerics. Ie. as it comes straight out of the interchange message. - :string_value: The single value string of the waveform data. :numeric_value: The - single value string of the waveform data. + exist). Exactly ONE of string_values, numeric_values must contain a non-None value. :return: True if write was successful. """ - num_non_nones = sum( - 1 for x in (values, string_value, numeric_value) if x is not None - ) + num_non_nones = sum(1 for x in (string_values, numeric_values) if x is not None) if num_non_nones != 1: raise ValueError( - "Exactly ONE of values, string_value, numeric_value must be not None" + "Exactly ONE of string_values, numeric_values must be not None" ) observation_datetime = datetime.fromtimestamp(observation_timestamp) @@ -67,12 +60,10 @@ def write_frame( ) filename.parent.mkdir(exist_ok=True, parents=True) - if values is not None: - # HF - csv_header = "csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,values\n" - else: - # LF - csv_header = "csn,mrn,source_variable_id,units,timestamp,location,string_value,numeric_value\n" + # The CSV fields are the same regardless of HF vs LF, to keep downstream + # processing simpler. Some fields may be nulled out, however. + # Single values will be wrapped in an array of length 1, if necessary. + csv_header = "csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,numeric_values,string_values\n" # write header if is new file if not filename.exists(): with open(filename, "w") as fileout: @@ -82,30 +73,18 @@ def write_frame( # predictable quoting makes testing easier wv_writer = csv.writer(fileout, delimiter=",", quoting=csv.QUOTE_ALL) - if values is not None: - # HF - row_array = [ - csn, - mrn, - source_variable_id, - source_channel_id, - units, - sampling_rate, - observation_timestamp, - mapped_location_string, - values, - ] - else: - row_array = [ - csn, - mrn, - source_variable_id, - units, - observation_timestamp, - mapped_location_string, - string_value if string_value is not None else "", - numeric_value if numeric_value is not None else "", - ] + row_array = [ + csn, + mrn, + source_variable_id, + source_channel_id if source_channel_id is not None else "", + units, + sampling_rate if sampling_rate is not None else "", + observation_timestamp, + mapped_location_string, + numeric_values if numeric_values is not None else "", + string_values if string_values is not None else "", + ] wv_writer.writerow(row_array) diff --git a/src/emap_interchange/messages.py b/src/emap_interchange/messages.py index 3163cdb..b9380a7 100644 --- a/src/emap_interchange/messages.py +++ b/src/emap_interchange/messages.py @@ -75,8 +75,8 @@ def get_sampling_rate(self): """Sampling rate in Hz.""" return self.data["samplingRate"] - def get_numeric_values(self): - """Numeric array.""" + def get_numeric_values(self) -> list[float]: + """Numeric array as a list of floats.""" return self.data["numericValues"]["value"] @@ -85,7 +85,7 @@ def get_source_value(self): """Unmapped value.""" return self.data["sourceValue"]["value"] - def get_numeric_value(self): + def get_numeric_value(self) -> float: """Mapped value, if it's a numerical value.""" return self.data["numericValue"]["value"] diff --git a/tests/test_controller.py b/tests/test_controller.py index 2c4acd3..003eb2d 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -52,7 +52,8 @@ def get_expected_write_frame_kwargs(self) -> dict: """If not bad data, what should write_frame be called with?""" fd = self.fake_data return { - "values": fd["numericValues"]["value"], + "numeric_values": fd["numericValues"]["value"], + "string_values": None, # HF is always numeric "source_variable_id": fd["sourceVariableId"], "source_channel_id": fd["sourceChannelId"], "observation_timestamp": fd["observationTime"], @@ -112,17 +113,22 @@ def get_expected_write_frame_kwargs(self) -> dict: fd = self.fake_data expected = { "source_variable_id": fd["sourceVariableId"], + "source_channel_id": None, "observation_timestamp": fd["observationTime"], "units": fd["unit"], + "sampling_rate": None, "mapped_location_string": fd["mappedLocationString"], "csn": "csn", "mrn": "mrn", } - if self.value_type == "numeric": - expected["numeric_value"] = fd["numericValue"]["value"] - elif self.value_type == "string": - expected["string_value"] = fd["stringValue"]["value"] + # LF may be string or numeric + expected["numeric_values"] = ( + [fd["numericValue"]["value"]] if self.value_type == "numeric" else None + ) + expected["string_values"] = ( + [fd["stringValue"]["value"]] if self.value_type == "string" else None + ) return expected @@ -174,8 +180,10 @@ def test_controller_callback( fake_data = fake_data_obj.get_fake_data() match bad_data_type: case 2: + # message type field missing del fake_data["@class"] case 3: + # message type field present but unrecognised fake_data["@class"] = fake_data["@class"].replace("e", "x") fake_data_str = json.dumps(fake_data) controller = WaveformController() diff --git a/tests/test_file_writer.py b/tests/test_file_writer.py index de0ad81..dafb898 100644 --- a/tests/test_file_writer.py +++ b/tests/test_file_writer.py @@ -59,24 +59,26 @@ def test_create_csv_low_freq( csn = "12345678" mrn = "whatever" - # it creates a separate file for each unit, so this becomes quite messy - expected_header = "csn,mrn,source_variable_id,units,timestamp,location,string_value,numeric_value\n" + # it creates a separate file for each unit, so we needs an expected text for each unit + expected_header = "csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,numeric_values,string_values\n" expected_texts = dict.fromkeys(units, expected_header) for idx, u in enumerate(units): v = values[idx] + string_values = numeric_values = None if type(v) is str: - value_kwarg = {"string_value": v} - expected_v_str = v + string_values = [v] + expected_v_str = f"['{v}']" expected_v_num = "" elif type(v) is float or type(v) is int: - value_kwarg = {"numeric_value": v} + numeric_values = [v] expected_v_str = "" - expected_v_num = str(v) + expected_v_num = f"[{v}]" else: raise ValueError(v) csv_writer.write_frame( - **value_kwarg, + string_values=string_values, + numeric_values=numeric_values, source_variable_id=variable_id, observation_timestamp=observation_time.timestamp(), units=u, @@ -84,14 +86,11 @@ def test_create_csv_low_freq( csn=csn, mrn=mrn, ) - expected_line = ( - f'"12345678","whatever","{variable_id}","{u}","1735726210.0",' - f'"mapped loc","{expected_v_str}","{expected_v_num}"\n' - ) + expected_line = f'"12345678","whatever","{variable_id}","","{u}","","1735726210.0","mapped loc","{expected_v_num}","{expected_v_str}"\n' expected_texts[u] += expected_line - # need to check all files now + # need to check multiple files for multiple units for idx, ef in enumerate(expected_filenames): - _check_write_csv(ef, expected_texts[units[idx]]) + _check_written_csv(ef, expected_texts[units[idx]]) @pytest.mark.parametrize( @@ -118,7 +117,7 @@ def test_create_csv_high_freq( mrn = "whatever" csv_writer.write_frame( - values="[1,2,3]", + numeric_values=[1, 2, 3.0], source_variable_id=variable_id, source_channel_id=channel_id, observation_timestamp=observation_time.timestamp(), @@ -130,10 +129,10 @@ def test_create_csv_high_freq( ) expected_text = ( - 'csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,values\n' - f'"12345678","whatever","{variable_id}","{channel_id or ""}","{units}","50","1735726210.0","mapped loc","[1,2,3]"\n' + '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' ) - _check_write_csv(expected_filename, expected_text) + _check_written_csv(expected_filename, expected_text) def _setup_write_csv(monkeypatch, tmp_path): @@ -147,7 +146,7 @@ def _setup_write_csv(monkeypatch, tmp_path): original_csv_dir.parent.mkdir(parents=True, exist_ok=True) -def _check_write_csv(expected_filename, expected_text): +def _check_written_csv(expected_filename, expected_text): # check that we can find the data again in its expected place expected_csv_path = locations.WAVEFORM_ORIGINAL_CSV / expected_filename assert os.path.exists(expected_csv_path) From 8b9496bf2a4df60680766cd20af77fdb7c7107c9 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Sun, 16 Aug 2026 23:16:06 +0100 Subject: [PATCH 06/13] Use JSON arrays in CSV --- src/csv_writer.py | 7 +++++-- tests/test_file_writer.py | 2 +- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/csv_writer.py b/src/csv_writer.py index 96aacdf..82a5e1b 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -1,6 +1,7 @@ """Writes a frame of waveform data to a csv file.""" import csv +import json from datetime import datetime from typing import Optional @@ -73,6 +74,8 @@ def write_frame( # predictable quoting makes testing easier wv_writer = csv.writer(fileout, delimiter=",", quoting=csv.QUOTE_ALL) + # Encode value lists as JSON so parquet conversion can use json.loads + # (Python list repr breaks on commas / quotes in string values). row_array = [ csn, mrn, @@ -82,8 +85,8 @@ def write_frame( sampling_rate if sampling_rate is not None else "", observation_timestamp, mapped_location_string, - numeric_values if numeric_values is not None else "", - string_values if string_values is not None else "", + json.dumps(numeric_values) if numeric_values is not None else "", + json.dumps(string_values) if string_values is not None else "", ] wv_writer.writerow(row_array) diff --git a/tests/test_file_writer.py b/tests/test_file_writer.py index dafb898..db5ba63 100644 --- a/tests/test_file_writer.py +++ b/tests/test_file_writer.py @@ -67,7 +67,7 @@ def test_create_csv_low_freq( string_values = numeric_values = None if type(v) is str: string_values = [v] - expected_v_str = f"['{v}']" + expected_v_str = f'[""{v}""]' expected_v_num = "" elif type(v) is float or type(v) is int: numeric_values = [v] From 40430604deeb4adf111836aab50e64dea1f89abe Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Mon, 17 Aug 2026 10:41:46 +0100 Subject: [PATCH 07/13] Extend LF data into parquet --- src/pseudon/pseudon.py | 42 ++++++--- tests/helpers.py | 78 ++++++++++++++-- tests/test_snakemake_integration.py | 139 ++++++++++++++++++++++------ 3 files changed, 211 insertions(+), 48 deletions(-) diff --git a/src/pseudon/pseudon.py b/src/pseudon/pseudon.py index 01cc498..628a136 100644 --- a/src/pseudon/pseudon.py +++ b/src/pseudon/pseudon.py @@ -4,7 +4,7 @@ import logging from decimal import Decimal from pathlib import Path -from typing import Any +from typing import Any, Optional import pandas as pd import pyarrow as pa @@ -19,6 +19,25 @@ from .hashing import do_hash +def _is_missing_csv_cell(x: Any) -> bool: + return x is None or x == "" or pd.isna(x) + + +def parse_numeric_values(x: Any) -> Optional[list[Decimal]]: + """Parse a JSON array of numbers from CSV into Decimals; empty cell -> None.""" + if _is_missing_csv_cell(x): + return None + # Not sure if this is the most efficient way. Might be able to do something with DecimalArray? + return list(json.loads(x, parse_float=Decimal, parse_int=Decimal)) + + +def parse_string_values(x: Any) -> Optional[list[str]]: + """Parse a JSON array of strings from CSV; empty cell -> None.""" + if _is_missing_csv_cell(x): + return None + return [str(i) for i in json.loads(x)] + + def pseudon_cli(): arg_parser = argparse.ArgumentParser() arg_parser.add_argument("--csv", type=Path) @@ -77,6 +96,8 @@ def csv_to_parquets( logger.info("Turning CSV %s to parquets", csv_path) csv_path.parent.mkdir(parents=True, exist_ok=True) original_parquet_path.parent.mkdir(parents=True, exist_ok=True) + # sampling_rate is null in low-frequency rows. Value columns are + # JSON arrays; exactly one of numeric_values / string_values is populated per row. df = pd.read_csv( str(csv_path), dtype={ @@ -85,20 +106,17 @@ def csv_to_parquets( "source_variable_id": str, "source_channel_id": str, "units": str, - "sampling_rate": int, + "sampling_rate": "Int32", "timestamp": float, "location": str, - "values": str, + "numeric_values": str, + "string_values": str, }, header=0, # the first line is always the header ) - def parse_array(x): - # Not sure if this is the most efficient way. Might be able to do something with DecimalArray? - # return [pa.decimal128(i) for i in x.replace(' ', '').split(',')] - return [Decimal(i) for i in x.strip().strip("[]").replace(" ", "").split(",")] - - df["values"] = df["values"].apply(parse_array) + df["numeric_values"] = df["numeric_values"].apply(parse_numeric_values) + df["string_values"] = df["string_values"].apply(parse_string_values) # Convert pandas DataFrame to pyarrow Table with proper types schema = pa.schema( @@ -119,7 +137,8 @@ def parse_array(x): # Not yet tested whether the specified precision # and scale cause it to be equivalent in size to decimal32. # See issue #31. - ("values", pa.list_(pa.decimal128(9, 4))), + ("numeric_values", pa.list_(pa.decimal128(9, 4))), + ("string_values", pa.list_(pa.string())), ] ) table = pa.Table.from_pandas(df, schema=schema, preserve_index=True) @@ -195,7 +214,8 @@ def add_waveform_metadata_to_table( "source_channel_id", "timestamp", "units", - "values", + "numeric_values", + "string_values", ] diff --git a/tests/helpers.py b/tests/helpers.py index 63d82f2..0ceaff0 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -2,8 +2,12 @@ from dataclasses import dataclass from decimal import Decimal import random +from typing import Literal, Optional + from stablehash import stablehash +ValueKind = Literal["numeric", "string"] + @dataclass class TestFileDescription: @@ -14,11 +18,25 @@ class TestFileDescription: mrn: str location: str variable_id: str - channel_id: str - sampling_rate: int + channel_id: Optional[str] + # None => low-frequency (no sampling rate in CSV) + sampling_rate: Optional[int] units: str num_rows: int - _test_values: list = None + # Which values column is populated. String values are LF-only in these fixtures. + value_kind: ValueKind = "numeric" + _test_numeric_values: Optional[list] = None + _test_string_values: Optional[list] = None + + def __post_init__(self): + if self.value_kind == "string" and self.sampling_rate is not None: + raise ValueError( + "string value_kind is only supported for low-frequency data" + ) + + @property + def is_low_freq(self) -> bool: + return self.sampling_rate is None def get_hashed_csn(self): """This test runs outside of a docker container and the hasher doesn't expose a @@ -64,6 +82,7 @@ def get_stable_hash(self): self.location, self.variable_id, self.channel_id, + self.value_kind, ) ) @@ -71,13 +90,29 @@ def get_stable_seed(self): byte_hash = self.get_stable_hash().digest()[:4] return int.from_bytes(byte_hash) - def generate_data(self, vals_per_row: int) -> list[list[Decimal]]: - if self._test_values is None: + def generate_data(self) -> tuple[list, list]: + """Return (numeric_rows, string_rows), each of length num_rows. + + Exactly one of the two lists has non-None row entries (lists of values); the + other is all None. CSV serialization of None is handled by write_frame. + """ + if self.value_kind == "string": + string_rows = self._generate_lf_string_rows() + return [None] * self.num_rows, string_rows + if self.is_low_freq: + numeric_rows = self._generate_lf_numeric_rows() + return numeric_rows, [None] * self.num_rows + vals_per_row = self.sampling_rate # one second of samples per row + numeric_rows = self._generate_hf_numeric_rows(vals_per_row) + return numeric_rows, [None] * self.num_rows + + def _generate_hf_numeric_rows(self, vals_per_row: int) -> list[list[Decimal]]: + if self._test_numeric_values is None: seed = self.get_stable_seed() rng = random.Random(seed) base_ampl = rng.normalvariate(1, 0.2) base_offset = rng.normalvariate(0, 0.2) - self._test_values = [] + self._test_numeric_values = [] for row_num in range(self.num_rows): values_row = [ Decimal.from_float( @@ -85,6 +120,31 @@ def generate_data(self, vals_per_row: int) -> list[list[Decimal]]: ).quantize(Decimal("1.0000")) for i in range(vals_per_row) ] - self._test_values.append(values_row) - # return as string but keep the numerical representation for comparison to parquet later - return self._test_values + self._test_numeric_values.append(values_row) + return self._test_numeric_values + + def _generate_lf_numeric_rows(self) -> list[list[Decimal]]: + if self._test_numeric_values is None: + seed = self.get_stable_seed() + rng = random.Random(seed) + self._test_numeric_values = [ + [Decimal.from_float(rng.uniform(0, 100)).quantize(Decimal("1.0000"))] + for _ in range(self.num_rows) + ] + return self._test_numeric_values + + def _generate_lf_string_rows(self) -> list[list[str]]: + if self._test_string_values is None: + # Realistic categorical / free-text LF examples (incl. commas / colons) + catalogue = [ + "Pressure Support / CPAP (PS)", + "1:2", + "Assist Control, Volume", + "SIMV", + ] + seed = self.get_stable_seed() + rng = random.Random(seed) + self._test_string_values = [ + [rng.choice(catalogue)] for _ in range(self.num_rows) + ] + return self._test_string_values diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 50d7b9c..42c9672 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -10,6 +10,8 @@ from pathlib import Path import pytest + +from src import csv_writer from tests.helpers import TestFileDescription @@ -30,7 +32,8 @@ def _run_compose( "sampling_rate", "timestamp", "location", - "values", + "numeric_values", + "string_values", ] REPO_ROOT = Path(__file__).resolve().parents[1] @@ -49,27 +52,66 @@ def build_required_images(): result.check_returncode() -def _make_test_input_csv(tmp_path, t: TestFileDescription) -> list[list[Decimal]]: +def _numeric_rows_as_written(numeric_rows: list) -> list: + """Expected parquet numerics after write_frame's json.dumps(float) round-trip.""" + result = [] + for row in numeric_rows: + if row is None: + result.append(None) + else: + as_floats = [float(v) for v in row] + result.append( + list( + json.loads( + json.dumps(as_floats), parse_float=Decimal, parse_int=Decimal + ) + ) + ) + return result + + +def _make_test_input_csv( + monkeypatch, tmp_path, t: TestFileDescription +) -> tuple[list, list]: + """Write CSV via csv_writer.write_frame (same path as production). + + Returns (numeric_rows, string_rows) expected in the resulting parquet. + """ + # Host tmp_path is mounted as /waveform-export in the exporter container, so + # CSVs must land at tmp_path/original-csv/... original_csv_dir = tmp_path / "original-csv" + original_csv_dir.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(csv_writer, "WAVEFORM_ORIGINAL_CSV", original_csv_dir) + + numeric_rows, string_rows = t.generate_data() + # Filename uses "noCh"; production write_frame gets None for missing channel. + 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): + csv_writer.write_frame( + numeric_values=( + [float(v) for v in numeric_values] + if numeric_values is not None + else None + ), + string_values=string_values, + source_variable_id=t.variable_id, + source_channel_id=source_channel_id, + observation_timestamp=row_time, + units=t.units, + sampling_rate=t.sampling_rate, + mapped_location_string=t.location, + csn=t.csn, + mrn=t.mrn, + ) + row_time += 1 # one second per row + csv_path = original_csv_dir / t.get_orig_csv() - csv_path.parent.mkdir(parents=True, exist_ok=True) - secs_per_row = 1 - vals_per_row = t.sampling_rate * secs_per_row - test_data = t.generate_data(vals_per_row) - with open(csv_path, "w") as f: - f.write(",".join(EXPECTED_COLUMN_NAMES) + "\n") - start_time = t.start_timestamp - row_time = start_time - for td in test_data: - row_values_str = ", ".join(str(v) for v in td) - f.write( - f'{t.csn},{t.mrn},{t.variable_id},{t.channel_id},{t.units},{t.sampling_rate},{row_time},{t.location},"[{row_values_str}]"\n' - ) - row_time += secs_per_row - # The test input CSV file needs to be old enough so that snakemake doesn't skip it + assert csv_path.exists(), f"write_frame did not create expected path {csv_path}" + # Old enough that snakemake does not skip it as "too new" old_time = time.time() - (10 * 60) os.utime(csv_path, (old_time, old_time)) - return test_data + return _numeric_rows_as_written(numeric_rows), string_rows @pytest.fixture(scope="function") @@ -107,7 +149,7 @@ def background_hasher(): ).check_returncode() -def test_snakemake_pipeline(tmp_path: Path, background_hasher): +def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): # ARRANGE # all fields that need to be de-IDed should contain the string "SECRET" so we can search for it later @@ -162,10 +204,38 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher): "uV", 5, ) + # low-frequency numeric (timestamps kept inside existing CSN_1234 day-1 range) + file5_lf_numeric = TestFileDescription( + "2025-01-01", + 1735740770.0, + "SECRET_CSN_1234", + "SECRET_MRN_12345", + "SECRET_LOCATION_123", + "1408", + "noCh", + None, + "unitless", + 3, + value_kind="numeric", + ) + # low-frequency string (timestamps kept inside existing CSN_1235 day-1 range) + file6_lf_string = TestFileDescription( + "2025-01-01", + 1735740784.0, + "SECRET_CSN_1235", + "SECRET_MRN_12346", + "SECRET_LOCATION_123", + "584", + "noCh", + None, + "unitless", + 2, + value_kind="string", + ) test_data_files = [] - for f in [file1, file2, file3, file4]: - test_data_values = _make_test_input_csv(tmp_path, f) - test_data_files.append((f, test_data_values)) + for f in [file1, file2, file3, file4, file5_lf_numeric, file6_lf_string]: + expected_values = _make_test_input_csv(monkeypatch, tmp_path, f) + test_data_files.append((f, expected_values)) expected_hash_summaries = { "2025-01-01": [ @@ -231,7 +301,7 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher): assert expected_summary == actual_hash_lookup_data # check no extraneous files - expected_file_counts = {"2025-01-01": 3, "2025-01-02": 1} + expected_file_counts = {"2025-01-01": 5, "2025-01-02": 1} _assert_date_partitioned_files(tmp_path / "original-csv", expected_file_counts) _assert_date_partitioned_files(tmp_path / "original-parquet", expected_file_counts) _assert_date_partitioned_files(tmp_path / "pseudonymised", expected_file_counts) @@ -305,13 +375,26 @@ def _run_snakemake(tmp_path): result.check_returncode() -def _compare_original_parquet_to_expected(original_parquet: Path, expected_test_values): - # CSV should always match original parquet +def _compare_original_parquet_to_expected( + original_parquet: Path, expected_data: tuple[list, list] +): + expected_numeric, expected_string = expected_data orig_parquet_file = pq.ParquetFile(original_parquet) orig_reader = orig_parquet_file.read() - orig_all_values = orig_reader["values"].combine_chunks() - expected_pa = pa.array(expected_test_values, type=orig_all_values.type) - assert orig_all_values == expected_pa + + orig_numeric = orig_reader["numeric_values"].combine_chunks() + if all(v is None for v in expected_numeric): + assert orig_numeric.null_count == len(orig_numeric) + else: + expected_pa = pa.array(expected_numeric, type=orig_numeric.type) + assert orig_numeric == expected_pa + + orig_string = orig_reader["string_values"].combine_chunks() + if all(v is None for v in expected_string): + assert orig_string.null_count == len(orig_string) + else: + expected_pa = pa.array(expected_string, type=orig_string.type) + assert orig_string == expected_pa def _compare_parquets(original_parquet_path: Path, pseudon_parquet_path: Path): From 5912b4ff62ce53be4236810e46d59712f8b27a4f Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Mon, 17 Aug 2026 16:48:22 +0100 Subject: [PATCH 08/13] Be consistent on line endings (LF everywhere) --- src/csv_writer.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/csv_writer.py b/src/csv_writer.py index 82a5e1b..3c27c13 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -67,12 +67,15 @@ def write_frame( csv_header = "csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,numeric_values,string_values\n" # write header if is new file if not filename.exists(): - with open(filename, "w") as fileout: + with open(filename, "w", newline="") as fileout: fileout.write(csv_header) - with open(filename, "a") as fileout: + # open with newline="" as per csv.writer docs + with open(filename, "a", newline="") as fileout: # predictable quoting makes testing easier - wv_writer = csv.writer(fileout, delimiter=",", quoting=csv.QUOTE_ALL) + wv_writer = csv.writer( + fileout, delimiter=",", quoting=csv.QUOTE_ALL, lineterminator="\n" + ) # Encode value lists as JSON so parquet conversion can use json.loads # (Python list repr breaks on commas / quotes in string values). From 6ab262c6758d255490d5e1556d1adeacbc361d9c Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Mon, 17 Aug 2026 17:52:56 +0100 Subject: [PATCH 09/13] Sort parquets by timestamp and enable more stats to hopefully speed up filtering/searching --- src/pseudon/pseudon.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/pseudon/pseudon.py b/src/pseudon/pseudon.py index 628a136..8200a5e 100644 --- a/src/pseudon/pseudon.py +++ b/src/pseudon/pseudon.py @@ -118,6 +118,9 @@ def csv_to_parquets( df["numeric_values"] = df["numeric_values"].apply(parse_numeric_values) df["string_values"] = df["string_values"].apply(parse_string_values) + # CSV row order follows RabbitMQ arrival, which is not guaranteed chronological. + df = df.sort_values("timestamp", kind="mergesort").reset_index(drop=True) + # Convert pandas DataFrame to pyarrow Table with proper types schema = pa.schema( [ @@ -154,7 +157,8 @@ def csv_to_parquets( # valid values: {‘NONE’, ‘SNAPPY’, ‘GZIP’, ‘BROTLI’, ‘LZ4’, ‘ZSTD’} compression="zstd", use_dictionary=True, - write_statistics=True, # enable indexes/statistics + write_statistics=True, + write_page_index=True, flavor="spark", ) logger.info( @@ -181,7 +185,8 @@ def csv_to_parquets( str(hashed_path), compression="zstd", use_dictionary=True, - write_statistics=True, # enable indexes/statistics + write_statistics=True, + write_page_index=True, flavor="spark", ) logger.info( From 7d89e4606c8e312dc888388b781ac0578f3e8a96 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Mon, 17 Aug 2026 17:58:22 +0100 Subject: [PATCH 10/13] Make some timestamps fractional so we can be sure there's no rounding going on --- tests/test_snakemake_integration.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 42c9672..92f6725 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -153,9 +153,10 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): # ARRANGE # all fields that need to be de-IDed should contain the string "SECRET" so we can search for it later + # Fractional seconds to ensure there's no integer rounding going on. file1 = TestFileDescription( "2025-01-01", - 1735740780.0, + 1735740780.25, "SECRET_CSN_1234", "SECRET_MRN_12345", "SECRET_LOCATION_123", @@ -181,7 +182,7 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): # same day, different CSN file3 = TestFileDescription( "2025-01-01", - 1735740783.0, + 1735740783.25, "SECRET_CSN_1235", "SECRET_MRN_12346", "SECRET_LOCATION_123", @@ -194,7 +195,7 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): # new day, first CSN again file4 = TestFileDescription( "2025-01-02", - 1735801965.0, + 1735801965, "SECRET_CSN_1234", "SECRET_MRN_12345", "SECRET_LOCATION_123", @@ -207,7 +208,7 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): # low-frequency numeric (timestamps kept inside existing CSN_1234 day-1 range) file5_lf_numeric = TestFileDescription( "2025-01-01", - 1735740770.0, + 1735740770.25, "SECRET_CSN_1234", "SECRET_MRN_12345", "SECRET_LOCATION_123", @@ -221,7 +222,7 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): # low-frequency string (timestamps kept inside existing CSN_1235 day-1 range) file6_lf_string = TestFileDescription( "2025-01-01", - 1735740784.0, + 1735740784.25, "SECRET_CSN_1235", "SECRET_MRN_12346", "SECRET_LOCATION_123", From 6d67734573f2934809c38f2ca978190df9d088ab Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Wed, 19 Aug 2026 11:16:05 +0100 Subject: [PATCH 11/13] Upgrade uv in the Docker image --- Dockerfile | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index e2fd34a..6393440 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,7 +6,8 @@ RUN export DEBIAN_FRONTEND=noninteractive && \ apt-get update && \ apt-get install --yes --no-install-recommends cron && \ apt-get autoremove --yes && apt-get clean --yes && rm -rf /var/lib/apt/lists/* -COPY --from=ghcr.io/astral-sh/uv@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4 /uv /uvx /bin/ +# uv image label "0.12.5" +COPY --from=ghcr.io/astral-sh/uv@sha256:e85be844203885286c60ffad8a858d48afb6c5a5c237ca0e67f12e74b8f174b1 /uv /uvx /bin/ ARG UVCACHE=/root/.cache/uv COPY PIXL /PIXL WORKDIR /app From a6995cd49b319dff89f6f79c21f5db7f153abd25 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Wed, 19 Aug 2026 11:16:55 +0100 Subject: [PATCH 12/13] Reject messages with both string and numeric values --- src/controller.py | 7 +++++++ tests/test_controller.py | 29 +++++++++++++++++++---------- 2 files changed, 26 insertions(+), 10 deletions(-) diff --git a/src/controller.py b/src/controller.py index e54dc98..8a9f1e2 100644 --- a/src/controller.py +++ b/src/controller.py @@ -107,6 +107,13 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): ) return + if (numeric_values is None) == (string_values is None): + reject_message(ch, method_frame.delivery_tag, False) + logger.error( + f"Waveform message {method_frame.delivery_tag} has either both numeric and string values, or neither." + ) + return + observation_time = datetime.fromtimestamp( observation_timestamp, tz=timezone.utc ) diff --git a/tests/test_controller.py b/tests/test_controller.py index 003eb2d..b2585f1 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1,6 +1,7 @@ import copy import json from datetime import datetime +from typing import Literal from unittest.mock import Mock import pytest @@ -15,7 +16,7 @@ class FakeData: def __init__(self, missing_key, value_type): self.missing_key = missing_key self.fake_data = FakeData._base_fake_data() - self.value_type = value_type + self.value_type: Literal["numeric", "string", "both"] = value_type @staticmethod def _base_fake_data() -> dict: @@ -76,24 +77,25 @@ def get_fake_data(self) -> dict: "value": None, "status": "IGNORE", } - if self.value_type == "numeric": + source_val = None + if self.value_type in ["numeric", "both"]: source_val = "0.8" self.fake_data["numericValue"] = { "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", "value": 0.8, "status": "SAVE", } - self.fake_data["stringValue"] = copy.copy(ignore_val) - elif self.value_type == "string": + if self.value_type in ["string", "both"]: source_val = "some categorical" self.fake_data["stringValue"] = { "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", "value": source_val, "status": "SAVE", } + if self.value_type == "numeric": + self.fake_data["stringValue"] = copy.copy(ignore_val) + elif self.value_type == "string": self.fake_data["numericValue"] = copy.copy(ignore_val) - else: - raise ValueError("must be numeric or string") self.fake_data["sourceValue"] = { "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue", "value": source_val, @@ -133,9 +135,10 @@ def get_expected_write_frame_kwargs(self) -> dict: @pytest.mark.parametrize( - # only affect LF tests so is redundant for HF + # only affect LF tests so is redundant for HF (which is always numeric) + # "both" is a form of bad data which we should reject "lf_value_type", - ["string", "numeric"], + ["string", "numeric", "both"], ) @pytest.mark.parametrize( "fake_data_class", @@ -162,6 +165,10 @@ def test_controller_callback( db_connect_failure, bad_data_type, ): + # Certain combinations of test just don't make sense + if fake_data_class == FakeHFData and lf_value_type != "numeric": + pytest.skip() + emap_db_mock = Mock() if db_connect_failure: emap_db_mock.get_row.side_effect = ConnectionError("mock database error") @@ -185,6 +192,7 @@ def test_controller_callback( case 3: # message type field present but unrecognised fake_data["@class"] = fake_data["@class"].replace("e", "x") + fake_data_str = json.dumps(fake_data) controller = WaveformController() @@ -196,11 +204,12 @@ def test_controller_callback( controller.waveform_callback(channel_mock, method_frame_mock, None, fake_data_str) - if not bad_data_type: + was_bad_data = bad_data_type or lf_value_type == "both" + if not was_bad_data: # we at least tried to query the DB emap_db_mock.get_row.assert_called_once() - if bad_data_type: + if was_bad_data: write_frame_mock.assert_not_called() # db should not even have been queried if data was bad emap_db_mock.get_row.assert_not_called() From 0f66a8c8cc106d979910eae8b3232ef0abfbb607 Mon Sep 17 00:00:00 2001 From: Jeremy Stein Date: Wed, 19 Aug 2026 11:43:23 +0100 Subject: [PATCH 13/13] Allow small drops in coverage (which is an artifact of the way we run tests, not a real drop in coverage anyway) --- codecov.yml | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 codecov.yml diff --git a/codecov.yml b/codecov.yml new file mode 100644 index 0000000..74ff384 --- /dev/null +++ b/codecov.yml @@ -0,0 +1,6 @@ +coverage: + status: + project: + default: + target: auto + threshold: 5%