From 4809a2e0e4b1f5881375583d0656ee51f85cf955 Mon Sep 17 00:00:00 2001 From: Ashlee Brunaugh Date: Fri, 7 Aug 2026 10:05:09 -0400 Subject: [PATCH 1/2] Harden PAQXOS frame validation --- docs/data_contracts.md | 10 +- docs/file_format.md | 23 ++- src/diffractomorph_pipeline/ingest.py | 92 +++++++-- src/diffractomorph_pipeline/io/paqxos_rtf.py | 17 +- src/diffractomorph_pipeline/study/manifest.py | 14 +- tests/public/test_ingest_validation.py | 180 ++++++++++++++++++ 6 files changed, 305 insertions(+), 31 deletions(-) create mode 100644 tests/public/test_ingest_validation.py diff --git a/docs/data_contracts.md b/docs/data_contracts.md index a54341c..50bb368 100644 --- a/docs/data_contracts.md +++ b/docs/data_contracts.md @@ -33,7 +33,8 @@ All adapters return `diffractomorph_pipeline.model.Run`, whose primary fields ar - `channel_ids`: stable, user-defined channel names; - `time_min`: chronological elapsed time; - `acquisition`: optional named frame-level variables; -- `stored_reference`: an optional static channel vector; +- `stored_reference`: an optional stored-reference channel vector, with its + selection and validation defined by the adapter; - `provenance`: explicit source, adapter, sample, and independent-unit identity. No fixed channel count, PAQXOS field, compound name, or preparation-date naming @@ -47,8 +48,11 @@ Adapters are selected explicitly by ID. Gate 1 includes: - `tidy_csv`: a vendor-neutral wide CSV with `time_min`, one or more `signal_` columns, optional `acq_` columns, and optional static `ref_` columns; -- `paqxos_rtf`: the existing Sympatec PAQXOS export reader mapped into the neutral - run model. +- `paqxos_rtf`: the Sympatec PAQXOS export reader mapped into the neutral run + model. When the instrument profile declares `channel_ids`, a frame is retained + only if it contains those identifiers exactly once, and output columns follow + the declared order. The adapter stores the earliest retained frame's reference + vector and flags reference variation across retained frames. Run `dfm-manifest --example --inspect-runs` to validate and inspect the packaged four-channel synthetic example without external data. diff --git a/docs/file_format.md b/docs/file_format.md index ff467b1..a7b1346 100644 --- a/docs/file_format.md +++ b/docs/file_format.md @@ -47,15 +47,21 @@ Channel, Ref Value, Measured Value | `Channel, Ref Value, Measured Value` | column header | | `, , ` | per detector ring: index, **reference**, **measured** | -## Three quirks the parser must respect +## Parser contracts 1. **Two intensity columns.** `Measured` is the per-frame signal (analysis target); `Ref` is the instrument's stored reference spectrum. **Capture both** — `Ref` feeds optional static-baseline subtraction (`I_bgsub = I − ref`) - and the staticness check. -2. **`Ref` is static within a file** — one fixed vector stamped into every frame - (per-channel std ≈ 1e-15). The parser validates this and emits a `ref_static` - flag; a varying `Ref` means a malformed/concatenated export. + and the reference-variation check. The adapter retains the earliest valid + frame's stored-reference vector and flags variation across retained frames; + variation does not automatically exclude a frame. +2. **Frame structure is validated against detector identifiers.** A PAQXOS + instrument profile may declare numeric `channel_ids` in strictly increasing + order. A frame is retained only when its timestamp is parseable and every + declared identifier occurs exactly once. Missing, additional, substituted, + or duplicate identifiers are counted by exclusion reason. Without declared + identifiers, the adapter infers a uniquely most frequent exact set and fails + closed when equally supported sets are ambiguous. 3. **Frames are listed newest-first** — raw document order is reverse-chronological. The parser **sorts by timestamp ascending** and records `reverse_order_detected`. (The legacy `diffractomorph_core` bug: it reversed @@ -70,7 +76,9 @@ The current parser treats each new `Measurement Time:` marker as the start of a sorts parsed timestamps chronologically. It does not depend on `@PAR(1)` as a structural separator. In the PAQXOS 5.1 manual, `@PAR(1)` means the first user-defined measurement parameter; it is retained in the archived template body for fidelity but is not required by -the parser or by intensity extraction. +the parser or by intensity extraction. Missing Copt, reference variation, +reverse document order, and interframe gaps are reported as run-level flags, +not additional automatic frame exclusions. ## Output contract @@ -83,7 +91,8 @@ compatibility properties. These names are not requirements for other adapters. `dfm-ingest` / `extract_run(emit_csv=True)` writes one tidy CSV per run — `frame, time_iso, t_min, copt, I_ch1 … I_ch31` — plus a `_meta.json` -sidecar carrying the static `ref` vector and `flags`. +sidecar carrying the earliest retained frame's stored-reference vector and +run-level flags. ## Open questions diff --git a/src/diffractomorph_pipeline/ingest.py b/src/diffractomorph_pipeline/ingest.py index c363bff..11a70d7 100644 --- a/src/diffractomorph_pipeline/ingest.py +++ b/src/diffractomorph_pipeline/ingest.py @@ -14,8 +14,8 @@ 1. **Two intensity columns per channel** — ``Ref Value`` and ``Measured Value``. ``Measured`` is the per-frame signal; ``Ref`` is the stored reference spectrum. Both are captured (``Ref`` feeds optional static-baseline subtraction). -2. **``Ref`` is static within a file** — validated; a varying ``Ref`` flags a - malformed/concatenated export. +2. **``Ref`` is expected to be static within a file** — variation is retained + and flagged for review rather than silently changing frame eligibility. 3. **Frames are listed newest-first** — the parser sorts by timestamp ascending and records ``reverse_order_detected``; document order is never trusted. @@ -26,8 +26,10 @@ import re import warnings +from collections import Counter from datetime import datetime from pathlib import Path +from typing import Sequence import numpy as np @@ -66,14 +68,17 @@ def _parse_frames(text: str) -> list[dict]: if cur is not None and cur["channels"]: frames.append(cur) cur = {"name": pending_name, "time": line.split(":", 1)[1].strip(), - "copt": float("nan"), "channels": {}} + "copt": float("nan"), "channels": {}, "duplicate_channels": set()} elif line.startswith("Optical Concentration:") and cur is not None: m = re.search(rf"({_NUM})", line.split(":", 1)[1]) cur["copt"] = float(m.group(1)) if m else float("nan") elif cur is not None: m = _CHANNEL_ROW.match(line) if m: - cur["channels"][int(m.group(1))] = (float(m.group(2)), float(m.group(3))) + channel = int(m.group(1)) + if channel in cur["channels"]: + cur["duplicate_channels"].add(channel) + cur["channels"][channel] = (float(m.group(2)), float(m.group(3))) if cur is not None and cur["channels"]: frames.append(cur) return frames @@ -86,6 +91,7 @@ def extract_run( run_kind: str | None = None, ref_static_tol: float = 1e-6, emit_csv: bool = False, + expected_channel_ids: Sequence[int | str] | None = None, ) -> RawRun: """Parse one PAQXOS RTF into a :class:`RawRun` (spec §4). @@ -99,6 +105,11 @@ def extract_run( Max per-channel std of ``ref`` across frames for the static check. emit_csv If true, also write the CSV mirror + meta sidecar next to the source. + expected_channel_ids + Detector-channel identifiers declared by the instrument profile. When + supplied, output columns follow this declared order. When omitted, the + parser infers the unique most frequently observed exact channel set and + fails closed if equally supported sets are ambiguous. """ from striprtf.striprtf import rtf_to_text @@ -106,23 +117,66 @@ def extract_run( text = rtf_to_text(path.read_text()) frames = _parse_frames(text) - # Canonical channel set = the channel list of the most complete frame. if not frames: raise ValueError(f"No frames parsed from {path}") - canonical = sorted(max(frames, key=lambda fr: len(fr["channels"]))["channels"].keys()) - n_ch = len(canonical) - - # Drop malformed frames (wrong channel count); fail only if <2 remain (§8). - good = [fr for fr in frames if sorted(fr["channels"].keys()) == canonical] - dropped = len(frames) - len(good) - if len(good) < 2 and len(frames) >= 2: - # too many dropped to trust — but a genuine single-frame file is allowed below - good = [fr for fr in frames if len(fr["channels"]) == n_ch] or good - if len(good) < 1: - raise ValueError(f"{path}: no valid frames after channel-count filtering") + + for frame in frames: + try: + frame["parsed_time"] = datetime.strptime(frame["time"], _TIME_FMT) + except ValueError: + frame["parsed_time"] = None + + if expected_channel_ids is not None: + try: + canonical = [int(str(channel)) for channel in expected_channel_ids] + except ValueError as exc: + raise ValueError("expected_channel_ids must contain integer identifiers") from exc + if not canonical: + raise ValueError("expected_channel_ids must not be empty") + if len(set(canonical)) != len(canonical): + raise ValueError("expected_channel_ids must be unique") + if canonical != sorted(canonical): + raise ValueError("expected_channel_ids must be in strictly increasing order") + channel_source = "argument" + else: + eligible_sets = [ + tuple(sorted(frame["channels"])) + for frame in frames + if not frame["duplicate_channels"] and frame["parsed_time"] is not None + ] + if not eligible_sets: + raise ValueError(f"{path}: no structurally valid frames from which to infer channels") + counts = Counter(eligible_sets) + top_count = max(counts.values()) + leaders = [channel_set for channel_set, count in counts.items() if count == top_count] + if len(leaders) != 1: + rendered = ", ".join(str(list(channel_set)) for channel_set in sorted(leaders)) + raise ValueError( + f"{path}: ambiguous detector-channel sets ({rendered}); " + "declare expected_channel_ids in the instrument profile" + ) + canonical = list(leaders[0]) + channel_source = "inferred" + + expected_set = set(canonical) + dropped_reasons: Counter[str] = Counter() + good = [] + for frame in frames: + if frame["duplicate_channels"]: + dropped_reasons["duplicate_channel_ids"] += 1 + elif frame["parsed_time"] is None: + dropped_reasons["unparseable_timestamp"] += 1 + elif set(frame["channels"]) != expected_set: + dropped_reasons["channel_set_mismatch"] += 1 + else: + good.append(frame) + dropped = sum(dropped_reasons.values()) + if not good: + detail = ", ".join(f"{key}={value}" for key, value in sorted(dropped_reasons.items())) + raise ValueError(f"{path}: no structurally valid frames ({detail})") # Document-order timestamps → detect reverse ordering, then sort ascending (§2.3). - times_doc = [datetime.strptime(fr["time"], _TIME_FMT) for fr in good] + times_doc = [fr["parsed_time"] for fr in good] reverse_order_detected = len(times_doc) > 1 and times_doc[0] > times_doc[-1] order = np.argsort(times_doc) good = [good[i] for i in order] @@ -131,7 +185,7 @@ def extract_run( I = np.array([[fr["channels"][c][1] for c in canonical] for fr in good], dtype=float) ref_all = np.array([[fr["channels"][c][0] for c in canonical] for fr in good], dtype=float) copt = np.array([fr["copt"] for fr in good], dtype=float) - times = [datetime.strptime(fr["time"], _TIME_FMT) for fr in good] + times = [fr["parsed_time"] for fr in good] t0 = times[0] t_min = np.array([(t - t0).total_seconds() / 60.0 for t in times]) @@ -158,6 +212,8 @@ def extract_run( "max_gap_min": max_gap_min, "n_frames": int(I.shape[0]), "dropped_frames": int(dropped), + "dropped_frame_reasons": dict(sorted(dropped_reasons.items())), + "expected_channel_ids_source": channel_source, "copt_nan": copt_nan, "run_kind_inferred": inferred, } diff --git a/src/diffractomorph_pipeline/io/paqxos_rtf.py b/src/diffractomorph_pipeline/io/paqxos_rtf.py index 49e64da..f487f71 100644 --- a/src/diffractomorph_pipeline/io/paqxos_rtf.py +++ b/src/diffractomorph_pipeline/io/paqxos_rtf.py @@ -9,7 +9,22 @@ class PaqxosRtfReader: adapter_id = "paqxos_rtf" def read(self, spec) -> Run: - run = ingest.extract_run(spec.source, run_kind=spec.run_kind) + return self._read(spec) + + def read_with_instrument_profile(self, spec, parameters) -> Run: + """Read using structural expectations declared by the instrument profile.""" + expected = parameters.get("channel_ids") + run = self._read(spec, expected_channel_ids=expected) + if expected is not None: + run.flags["expected_channel_ids_source"] = "instrument_profile" + return run + + def _read(self, spec, expected_channel_ids=None) -> Run: + run = ingest.extract_run( + spec.source, + run_kind=spec.run_kind, + expected_channel_ids=expected_channel_ids, + ) run.provenance = RunProvenance( run_id=spec.run_id, source_path=str(spec.source), diff --git a/src/diffractomorph_pipeline/study/manifest.py b/src/diffractomorph_pipeline/study/manifest.py index 168196a..33c2a9e 100644 --- a/src/diffractomorph_pipeline/study/manifest.py +++ b/src/diffractomorph_pipeline/study/manifest.py @@ -212,16 +212,26 @@ def _validate_loaded_run(self, spec: RunSpec, run: Run) -> Run: ) return run + def _read_spec(self, spec: RunSpec) -> Run: + reader = get_reader(spec.adapter) + instrument = self.require_profile("instrument") + profile_aware_read = getattr(reader, "read_with_instrument_profile", None) + if profile_aware_read is not None: + run = profile_aware_read(spec, instrument.parameters) + else: + run = reader.read(spec) + return self._validate_loaded_run(spec, run) + def read_run(self, run_id: str) -> Run: """Read one run through the adapter explicitly declared in the manifest.""" matches = [spec for spec in self.runs if spec.run_id == run_id] if not matches: raise KeyError(f"unknown run_id {run_id!r}") spec = matches[0] - return self._validate_loaded_run(spec, get_reader(spec.adapter).read(spec)) + return self._read_spec(spec) def read_all_runs(self) -> tuple[Run, ...]: - return tuple(self._validate_loaded_run(spec, get_reader(spec.adapter).read(spec)) for spec in self.runs) + return tuple(self._read_spec(spec) for spec in self.runs) def bundled_example_manifest() -> Path: diff --git a/tests/public/test_ingest_validation.py b/tests/public/test_ingest_validation.py new file mode 100644 index 0000000..5722b4e --- /dev/null +++ b/tests/public/test_ingest_validation.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +from datetime import datetime, timedelta +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pytest + +from diffractomorph_pipeline.ingest import extract_run +from diffractomorph_pipeline.io.paqxos_rtf import PaqxosRtfReader +from diffractomorph_pipeline.study.manifest import ProfileSpec, ProjectManifest, RunSpec + + +def _write_rtf(path: Path, frames: list[dict], *, newest_first: bool = True) -> Path: + ordered = sorted(frames, key=lambda frame: frame["time"], reverse=newest_first) + lines: list[str] = [] + for frame in ordered: + lines.extend( + [ + "Measurement Name: synthetic validation run", + f"Measurement Time: {frame['time'].strftime('%Y-%m-%d %H:%M:%S')}", + "Optical Concentration: " + + ("" if frame.get("copt") is None else str(frame["copt"])), + "Channel, Ref Value, Measured Value", + ] + ) + for channel, reference, measured in frame["rows"]: + lines.append(f"{channel}, {reference}, {measured}") + path.write_text("{\\rtf1\\ansi\n" + "\\par\n".join(lines) + "\\par\n}") + return path + + +def _frame(index: int, channels=(1, 2, 3), *, copt=1.0) -> dict: + started = datetime(2026, 1, 1, 12, 0, 0) + return { + "time": started + timedelta(seconds=30 * index), + "copt": copt, + "rows": [(channel, 0.1 * channel, 10 * index + channel) for channel in channels], + } + + +def test_expected_channels_drop_same_length_wrong_identifier_set(tmp_path): + source = _write_rtf( + tmp_path / "run.rtf", + [_frame(0), _frame(1, channels=(1, 2, 4)), _frame(2)], + ) + + run = extract_run(source, run_kind="measurement", expected_channel_ids=(1, 2, 3)) + + assert run.channel_ids == ("1", "2", "3") + assert run.signal.shape == (2, 3) + assert run.flags["dropped_frames"] == 1 + assert run.flags["dropped_frame_reasons"] == {"channel_set_mismatch": 1} + + +def test_expected_channels_fail_when_no_frame_matches(tmp_path): + source = _write_rtf( + tmp_path / "run.rtf", + [_frame(0, channels=(1, 2, 4)), _frame(1, channels=(1, 2, 4))], + ) + + with pytest.raises(ValueError, match="no structurally valid frames"): + extract_run(source, expected_channel_ids=(1, 2, 3)) + + +def test_expected_channels_must_use_numeric_detector_order(tmp_path): + source = _write_rtf(tmp_path / "run.rtf", [_frame(0), _frame(1)]) + + with pytest.raises(ValueError, match="strictly increasing order"): + extract_run(source, expected_channel_ids=(2, 1, 3)) + + +def test_inference_fails_closed_for_equally_complete_channel_sets(tmp_path): + source = _write_rtf( + tmp_path / "run.rtf", + [_frame(0, channels=(1, 2, 3)), _frame(1, channels=(1, 2, 4))], + ) + + with pytest.raises(ValueError, match="ambiguous detector-channel sets"): + extract_run(source) + + +def test_duplicate_channel_row_is_not_silently_overwritten(tmp_path): + duplicate = _frame(1) + duplicate["rows"].append((2, 99.0, 99.0)) + source = _write_rtf(tmp_path / "run.rtf", [_frame(0), duplicate, _frame(2)]) + + run = extract_run(source, expected_channel_ids=(1, 2, 3)) + + assert run.signal.shape == (2, 3) + assert run.flags["dropped_frame_reasons"] == {"duplicate_channel_ids": 1} + + +def test_missing_copt_and_reference_variation_are_retained_as_flags(tmp_path): + frames = [_frame(0, copt=None), _frame(1)] + frames[1]["rows"][0] = (1, 0.5, 11.0) + source = _write_rtf(tmp_path / "run.rtf", frames) + + with pytest.warns(UserWarning) as recorded: + run = extract_run(source, expected_channel_ids=(1, 2, 3)) + + assert len(recorded) == 2 + assert run.flags["copt_nan"] == 1 + assert run.flags["ref_static"] is False + assert run.flags["dropped_frames"] == 0 + assert np.isnan(run.acquisition["copt"][0]) + + +def test_profile_aware_reader_passes_declared_channels_to_parser(tmp_path): + source = _write_rtf( + tmp_path / "run.rtf", + [_frame(0), _frame(1, channels=(1, 2, 4)), _frame(2)], + ) + spec = SimpleNamespace( + source=source, + run_kind="measurement", + run_id="run-1", + sample_id="sample-1", + independent_unit_id="prep-1", + technical_replicate="1", + instrument_id="instrument-1", + metadata={}, + ) + + run = PaqxosRtfReader().read_with_instrument_profile( + spec, {"adapter": "paqxos_rtf", "channel_ids": [1, 2, 3]} + ) + + assert run.signal.shape == (2, 3) + assert run.flags["expected_channel_ids_source"] == "instrument_profile" + + +def _project(source: Path, channel_ids) -> ProjectManifest: + instrument_parameters = {"adapter": "paqxos_rtf"} + if channel_ids is not None: + instrument_parameters["channel_ids"] = channel_ids + spec = RunSpec( + run_id="run-1", + source=source, + adapter="paqxos_rtf", + run_kind="measurement", + sample_id="sample-1", + independent_unit_id="prep-1", + ) + return ProjectManifest( + manifest_path=source.parent / "project.yaml", + schema_version=1, + project_id="validation-project", + data_root=source.parent, + independent_unit="preparation", + profiles={ + "instrument": ProfileSpec( + role="instrument", + profile_id="instrument-1", + parameters=instrument_parameters, + ) + }, + runs=(spec,), + ) + + +def test_manifest_declared_channels_fail_when_no_frame_matches(tmp_path): + source = _write_rtf( + tmp_path / "run.rtf", + [_frame(0, channels=(1, 2, 4)), _frame(1, channels=(1, 2, 4))], + ) + + with pytest.raises(ValueError, match="no structurally valid frames"): + _project(source, [1, 2, 3]).read_run("run-1") + + +def test_manifest_without_declared_channels_fails_on_ambiguous_sets(tmp_path): + source = _write_rtf( + tmp_path / "run.rtf", + [_frame(0, channels=(1, 2, 3)), _frame(1, channels=(1, 2, 4))], + ) + + with pytest.raises(ValueError, match="ambiguous detector-channel sets"): + _project(source, None).read_run("run-1") From 2c0fb3491c6331d018e7d4f38c8eb4ee08885274 Mon Sep 17 00:00:00 2001 From: Ashlee Brunaugh Date: Fri, 7 Aug 2026 10:07:21 -0400 Subject: [PATCH 2/2] Clarify PAQXOS frame-order contract --- docs/file_format.md | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/docs/file_format.md b/docs/file_format.md index a7b1346..232881f 100644 --- a/docs/file_format.md +++ b/docs/file_format.md @@ -62,11 +62,8 @@ Channel, Ref Value, Measured Value or duplicate identifiers are counted by exclusion reason. Without declared identifiers, the adapter infers a uniquely most frequent exact set and fails closed when equally supported sets are ambiguous. -3. **Frames are listed newest-first** — raw document order is - reverse-chronological. The parser **sorts by timestamp ascending** and records - `reverse_order_detected`. (The legacy `diffractomorph_core` bug: it reversed - the list on a wrong assumption and computed frame-to-frame metrics on - non-adjacent pairs.) +3. **Frames may be listed newest-first.** The parser sorts by timestamp ascending + and records whether reverse document order was detected. `Measured` is **not** background-subtracted: in a blank, `Measured ≈ Ref ≈ 0.3–0.6` (not zero). `Measured − Ref` is the drug-attributable