Allow low frequency waveform data into parquet

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

README.md

@@ -6,6 +6,8 @@
 
 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

codecov.yml

@@ -0,0 +1,6 @@
+coverage:
+  status:
+    project:
+      default:
+        target: auto
+        threshold: 5%

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

src/controller.py

@@ -3,13 +3,18 @@
 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,29 +53,67 @@ 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"]
-            logger.debug(
-                "Message is for loc %s, var %s, ch %s",
-                location_string,
-                source_variable_id,
-                source_channel_id,
-            )
+            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:
+            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):
+                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,
+                    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:
+                    string_values = [string_value]
+
+                numeric_value = message.get_numeric_value()
+                if numeric_value is not None:
+                    numeric_values = [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(
                 f"Waveform message {method_frame.delivery_tag} is missing required data {e}."
             )
             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
         )
@@ -98,15 +141,16 @@ 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,
+            sampling_rate=sampling_rate,
+            observation_timestamp=observation_timestamp,
+            units=units,
+            mapped_location_string=mapped_location_string,
+            csn=csn,
+            mrn=mrn,
+            numeric_values=numeric_values,
+            string_values=string_values,
         ):
             if lookup_success:
                 ack_message(ch, method_frame.delivery_tag)

src/csv_writer.py

@@ -1,14 +1,16 @@
 """Writes a frame of waveform data to a csv file."""
 
 import csv
+import json
 from datetime import datetime
+from typing import Optional
 
 from locations import WAVEFORM_ORIGINAL_CSV, make_file_name, FILE_STEM_PATTERN
 
 
 def create_file_name(
     source_variable_id: str,
-    source_channel_id: str,
+    source_channel_id: Optional[str],
     observation_time: datetime,
     csn: str,
     units: str,
@@ -30,50 +32,66 @@ def create_file_name(
 
 
 def write_frame(
-    waveform_data: dict,
+    *,
+    numeric_values: Optional[list[float]] = None,
+    string_values: Optional[list[str]] = 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.
+    """Appends a frame of waveform data to a csv file (creates file if it doesn't
+    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 (string_values, numeric_values) if x is not None)
+    if num_non_nones != 1:
+        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
     )
     filename.parent.mkdir(exist_ok=True, parents=True)
 
+    # 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:
-            fileout.write(
-                "csn,mrn,source_variable_id,source_channel_id,units,sampling_rate,timestamp,location,values\n"
-            )
-
-    with open(filename, "a") as fileout:
-        wv_writer = csv.writer(fileout, delimiter=",")
-        waveform_data = waveform_data.get("value", "")
+        with open(filename, "w", newline="") as fileout:
+            fileout.write(csv_header)
 
-        wv_writer.writerow(
-            [
-                csn,
-                mrn,
-                source_variable_id,
-                source_channel_id,
-                units,
-                sampling_rate,
-                observation_timestamp,
-                mapped_location_string,
-                waveform_data,
-            ]
+    # 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, lineterminator="\n"
         )
 
+        # 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,
+            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,
+            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)
+
     return True

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) -> list[float]:
+        """Numeric array as a list of floats."""
+        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) -> float:
+        """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"]