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
diff --git a/README.md b/README.md
index 535cf1d..02f46ef 100644
--- a/README.md
+++ b/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
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%
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
diff --git a/src/controller.py b/src/controller.py
index 3dc3099..8a9f1e2 100644
--- a/src/controller.py
+++ b/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,22 +53,53 @@ 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(
@@ -71,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
         )
@@ -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)
diff --git a/src/csv_writer.py b/src/csv_writer.py
index 66948b8..3c27c13 100644
--- a/src/csv_writer.py
+++ b/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,20 +32,28 @@ 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(
@@ -51,29 +61,37 @@ def write_frame(
     )
     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
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..b9380a7
--- /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) -> 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"]
diff --git a/src/pseudon/pseudon.py b/src/pseudon/pseudon.py
index 01cc498..8200a5e 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,20 @@ 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["numeric_values"] = df["numeric_values"].apply(parse_numeric_values)
+    df["string_values"] = df["string_values"].apply(parse_string_values)
 
-    df["values"] = df["values"].apply(parse_array)
+    # 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(
@@ -119,7 +140,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)
@@ -135,7 +157,8 @@ def parse_array(x):
         # 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(
@@ -162,7 +185,8 @@ def parse_array(x):
         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(
@@ -195,7 +219,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_controller.py b/tests/test_controller.py
index d4fd611..b2585f1 100644
--- a/tests/test_controller.py
+++ b/tests/test_controller.py
@@ -1,5 +1,7 @@
+import copy
 import json
 from datetime import datetime
+from typing import Literal
 from unittest.mock import Mock
 
 import pytest
@@ -7,6 +9,141 @@
 from controller import WaveformController
 
 
+class FakeData:
+    """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: Literal["numeric", "string", "both"] = value_type
+
+    @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.WaveformHighFreqMessage"
+        )
+        # simulate a missing key
+        if not self.missing_key:
+            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_kwargs(self) -> dict:
+        """If not bad data, what should write_frame be called with?"""
+        fd = self.fake_data
+        return {
+            "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"],
+            "units": fd["unit"],
+            "sampling_rate": fd["samplingRate"],
+            "mapped_location_string": fd["mappedLocationString"],
+            "csn": "csn",
+            "mrn": "mrn",
+        }
+
+
+class FakeLFData(FakeData):
+    def get_fake_data(self) -> dict:
+        self.fake_data["@class"] = (
+            "uk.ac.ucl.rits.inform.interchange.visit_observations.WaveformLowFreqMessage"
+        )
+
+        ignore_val = {
+            "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue",
+            "value": None,
+            "status": "IGNORE",
+        }
+        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",
+            }
+        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)
+        self.fake_data["sourceValue"] = {
+            "@class": "uk.ac.ucl.rits.inform.interchange.InterchangeValue",
+            "value": source_val,
+            "status": "SAVE",
+        }
+        if self.missing_key:
+            # simulate a missing key
+            del self.fake_data["sourceVariableId"]
+        return self.fake_data
+
+    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
+        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",
+        }
+
+        # 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
+
+
+@pytest.mark.parametrize(
+    # 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", "both"],
+)
+@pytest.mark.parametrize(
+    "fake_data_class",
+    [FakeHFData, FakeLFData],
+)
 @pytest.mark.parametrize(
     "opt_out",
     [True, False],
@@ -16,10 +153,22 @@
     [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, opt_out, db_connect_failure, bad_data):
+def test_controller_callback(
+    monkeypatch,
+    lf_value_type,
+    fake_data_class,
+    opt_out,
+    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")
@@ -30,19 +179,20 @@ 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"]
+    # 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), value_type=lf_value_type
+    )
+    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()
 
@@ -54,11 +204,12 @@ def test_controller_callback(monkeypatch, opt_out, db_connect_failure, bad_data)
 
     controller.waveform_callback(channel_mock, method_frame_mock, None, fake_data_str)
 
-    if not bad_data:
+    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:
+    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()
@@ -76,6 +227,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_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..db5ba63 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,91 @@
 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 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:
+            string_values = [v]
+            expected_v_str = f'[""{v}""]'
+            expected_v_num = ""
+        elif type(v) is float or type(v) is int:
+            numeric_values = [v]
+            expected_v_str = ""
+            expected_v_num = f"[{v}]"
+        else:
+            raise ValueError(v)
+
+        csv_writer.write_frame(
+            string_values=string_values,
+            numeric_values=numeric_values,
+            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","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):
+        _check_written_csv(ef, expected_texts[units[idx]])
+
+
 @pytest.mark.parametrize(
     "units, variable_id, channel_id, expected_filename",
     [
@@ -16,9 +102,40 @@
         ("%", "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,
 ):
+    _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(
+        numeric_values=[1, 2, 3.0],
+        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,
+    )
+
+    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'
+    )
+    _check_written_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("/")
@@ -28,22 +145,10 @@ def test_create_file_name_handles_units(
     # the only precondition is that the base dir must exist
     original_csv_dir.parent.mkdir(parents=True, exist_ok=True)
 
-    observation_time = datetime(2025, 1, 1, 10, 10, 10, tzinfo=timezone.utc)
-    csn = "12345678"
-    mrn = "whatever"
-
-    csv_writer.write_frame(
-        {"value": "[1,2,3]"},
-        variable_id,
-        channel_id,
-        observation_time.timestamp(),
-        units,
-        50,
-        "mapped loc",
-        csn,
-        mrn,
-    )
 
+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)
+    actual_text = expected_csv_path.read_text()
+    assert actual_text == expected_text
diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py
index 50d7b9c..92f6725 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,13 +149,14 @@ 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
+    # 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",
@@ -139,7 +182,7 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher):
     # same day, different CSN
     file3 = TestFileDescription(
         "2025-01-01",
-        1735740783.0,
+        1735740783.25,
         "SECRET_CSN_1235",
         "SECRET_MRN_12346",
         "SECRET_LOCATION_123",
@@ -152,7 +195,7 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher):
     # new day, first CSN again
     file4 = TestFileDescription(
         "2025-01-02",
-        1735801965.0,
+        1735801965,
         "SECRET_CSN_1234",
         "SECRET_MRN_12345",
         "SECRET_LOCATION_123",
@@ -162,10 +205,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.25,
+        "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.25,
+        "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 +302,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 +376,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):